From f0f6dba8cf637bdc9308ff1d513103d673a54fed Mon Sep 17 00:00:00 2001 From: kahlun Date: Thu, 24 Sep 2026 02:06:46 +0000 Subject: [PATCH 01/11] experiment: XPU reduce_avg + attention support, pure monkeypatch Split out of experiment/xpu-monkeypatch-no-core-hooks: covers is_reduce_avg_supported (FSDP2 per-layer loop, no patch needed; raw dist.all_reduce(AVG), patched) and attention_utils_module. Both work with zero verl-core changes, independent of the VTune profiler half (docs/design/xpu-monkeypatch-vtune-profiler.md), which still has an open blocker. Co-Authored-By: Claude Sonnet 5 --- .../xpu-monkeypatch-reduce-avg-attention.md | 80 +++++++++++++++++++ verl_hardware_plugin/__init__.py | 2 + verl_hardware_plugin/engines/fsdp_xpu.py | 51 ++++++++++-- verl_hardware_plugin/patches/__init__.py | 40 ++++++++++ .../patches/attention_patch_xpu.py | 55 +++++++++++++ .../patches/reduce_avg_allreduce_patch_xpu.py | 72 +++++++++++++++++ 6 files changed, 292 insertions(+), 8 deletions(-) create mode 100644 docs/design/xpu-monkeypatch-reduce-avg-attention.md create mode 100644 verl_hardware_plugin/patches/__init__.py create mode 100644 verl_hardware_plugin/patches/attention_patch_xpu.py create mode 100644 verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py diff --git a/docs/design/xpu-monkeypatch-reduce-avg-attention.md b/docs/design/xpu-monkeypatch-reduce-avg-attention.md new file mode 100644 index 00000000..f5a81a6d --- /dev/null +++ b/docs/design/xpu-monkeypatch-reduce-avg-attention.md @@ -0,0 +1,80 @@ +# Experiment: reduce_avg + attention support for XPU, pure monkeypatch + +## Context + +`verl-hardware-plugin#26` asked maintainers whether Intel XPU should extend +`PlatformBase` in `verl-core` (the TPU precedent) or stay entirely inside +this plugin via monkeypatch (the MLU precedent). The maintainer's answer +(2026-09-23): default to the plugin, keep `verl-project/verl#7917` (which +adds `is_reduce_avg_supported`/`attention_utils_module`/`profiler_markers`/ +`dist_profiler_cls` to `PlatformBase`) narrowed to only what truly can't be +done from here. + +This branch covers two of those four capabilities — +`is_reduce_avg_supported` and `attention_utils_module` — split out from the +original combined experiment (`experiment/xpu-monkeypatch-no-core-hooks`) +so this half, which has no unresolved design smell beyond the one flagged +below, can be reviewed and hardware-validated independently of the VTune +profiler half (`docs/design/xpu-monkeypatch-vtune-profiler.md`), which has +an open blocker. + +Does not touch `feature/xpu-vtune-avg-attention` (the branch behind open +PR #22, which uses the hook-based approach and depends on #7917 merging). +Built from `main` so it can be compared directly for footprint and +core-change dependency. + +## Verdict per capability + +| Capability | Mechanism | Notes | +|---|---|---| +| `attention_utils_module` | Patch `verl.utils.attention_utils._get_attention_functions` | Clean. That function is re-invoked on every call by the public wrappers, so patch order doesn't matter. | +| `is_reduce_avg_supported` (FSDP2) | Not a patch at all — extended `verl_hardware_plugin/engines/fsdp_xpu.py`'s existing `initialize()` to loop over every FSDP2-wrapped submodule, not just the root | See below — this is *better* than the monkeypatch this branch originally planned, and matches the per-layer+root fix already validated on real B60 hardware via a 12-run controlled A/B (6/6 pass with the fix, 6/6 fail without, on `run_grpo_fsdp2_intel_gpu.sh`). | +| `is_reduce_avg_supported` (3 raw `dist.all_reduce(op=AVG)` sites) | Patch `torch.distributed.all_reduce` process-wide when `op=AVG` | Robust to import order (attribute lookup at call time), but changes global `ReduceOp.AVG` semantics for the whole process. Real design smell, documented in `patches/reduce_avg_allreduce_patch_xpu.py`. This is the one piece here still worth a narrow core-hook ask if the process-wide semantics change is judged unacceptable on review. | + +Net: both capabilities work today with zero `verl-core` changes. + +## Why the FSDP2 fix isn't a patch + +The first draft of this plan (matching the original proposal doc) was to +wrap `verl.utils.fsdp_utils.apply_fsdp2`: call the original, then walk +`model.modules()` and call `set_force_sum_reduction_for_comms(True)`. + +That doesn't actually work reliably: `verl/workers/engine/fsdp/ +transformer_impl.py` does `from verl.utils.fsdp_utils import apply_fsdp2` at +its own import time. Reassigning `verl.utils.fsdp_utils.apply_fsdp2` after +that has already happened has no effect on the name bound inside +`transformer_impl.py`. + +This plugin doesn't need that patch at all, though: `fsdp_xpu.py`'s engine +subclasses already call `set_force_sum_reduction_for_comms` on the +constructed model *after* `super().initialize()` returns — from inside their +own `EngineRegistry`-registered class, which verl-core calls directly (no +patching involved). This branch extends that existing, already-real +mechanism to loop over every FSDP2-wrapped submodule instead of only the +root (`_force_sum_reduction_on_all_fsdp_modules` in `fsdp_xpu.py`), achieved +here with no patch and no hook. + +Note: PR #22's current `fsdp_xpu.py` (the hook-based branch) only fixes a +`self.model`→`self.module` attribute bug and still calls +`set_force_sum_reduction_for_comms` on the root only — it does not yet +include this per-layer loop. Whichever branch ships, this fix should land; +it isn't specific to the monkeypatch-vs-hook question. + +## Comparison + +| | Files touched outside this plugin | Depends on #7917 merging | +|---|---|---| +| PR #22 (`feature/xpu-vtune-avg-attention`) | 0 (but the `is_reduce_avg_supported`/`attention_utils_module` hooks are dead code without #7917) | Yes | +| This branch | 0 | No | + +## Not yet done + +- No hardware validation of the `attention_utils_module` patch or the raw + `dist.all_reduce(AVG)` patch — reasoned through against `verl-core`'s real + `main` source, not run on a B60. The FSDP2 per-layer fix itself is already + hardware-validated (see the 12-run A/B cited above); only its + monkeypatch-vs-engine-override framing here is new. +- No unit tests yet for `patches/`. +- Whether to still ask for a narrow core hook covering the 3 raw + `dist.all_reduce(AVG)` call sites, given the process-wide semantics change + this patch makes, is an open question for review. diff --git a/verl_hardware_plugin/__init__.py b/verl_hardware_plugin/__init__.py index 90ee349b..90b7e4ee 100644 --- a/verl_hardware_plugin/__init__.py +++ b/verl_hardware_plugin/__init__.py @@ -14,6 +14,7 @@ import os from verl_hardware_plugin.engines import register_all_engines +from verl_hardware_plugin.patches import apply_all as apply_all_patches from verl_hardware_plugin.platforms import register_all_platforms from verl_hardware_plugin.profilers import register_all_profiles @@ -23,5 +24,6 @@ register_all_platforms() register_all_engines() register_all_profiles() +apply_all_patches() logger.info("verl-hardware-plugin loaded successfully") diff --git a/verl_hardware_plugin/engines/fsdp_xpu.py b/verl_hardware_plugin/engines/fsdp_xpu.py index 6c7397b1..95c0f1e7 100644 --- a/verl_hardware_plugin/engines/fsdp_xpu.py +++ b/verl_hardware_plugin/engines/fsdp_xpu.py @@ -13,6 +13,21 @@ followed by manual division, which is functionally equivalent but compatible with xccl. +Why not patch verl-core's apply_fsdp2() instead? + verl/workers/engine/fsdp/transformer_impl.py does + `from verl.utils.fsdp_utils import apply_fsdp2` at its own import time, so a + plugin that reassigns `verl.utils.fsdp_utils.apply_fsdp2` after that module + has already imported the name would have no effect on this call site -- + the same import-order fragility as verl core's profiler-marker dispatch + (see docs/design/ for that one). This engine sidesteps the problem + entirely: it does not patch anything. It calls + set_force_sum_reduction_for_comms() on the already-constructed model + *after* super().initialize() (which internally calls the real + apply_fsdp2()) returns, from inside the plugin's own EngineRegistry + subclass. That is ordinary method-override behavior, not a monkeypatch, + so it is robust regardless of when/how anything else imported + apply_fsdp2. + Registration: @EngineRegistry.register(device="xpu", vendor="intel") This means verl will automatically select this engine when: @@ -38,6 +53,28 @@ logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN")) +def _force_sum_reduction_on_all_fsdp_modules(root) -> None: + """Apply set_force_sum_reduction_for_comms(True) to root and every nested + FSDP2-wrapped submodule, not just the root. + + fully_shard() wraps the transformer layers individually as well as the + root model (verl/utils/fsdp_utils.py::apply_fsdp2), so each of those + per-layer wrappers has its own communication group that also defaults to + ReduceOp.AVG. Only fixing the root leaves per-layer gradient sync using + AVG, which xccl silently cannot execute correctly. + """ + if hasattr(root, "set_force_sum_reduction_for_comms"): + root.set_force_sum_reduction_for_comms(True) + count = 1 + for submodule in root.modules(): + if submodule is root: + continue + if hasattr(submodule, "set_force_sum_reduction_for_comms"): + submodule.set_force_sum_reduction_for_comms(True) + count += 1 + logger.info("Enabled force_sum_reduction_for_comms on %d FSDP module(s) for XPU", count) + + @EngineRegistry.register(model_type="language_model", backend=["fsdp", "fsdp2"], device="xpu", vendor="intel") class FSDPXPUEngineWithLMHead(FSDPEngineWithLMHead): """FSDP Engine for Intel XPU with xccl communication backend. @@ -59,15 +96,14 @@ def __init__( def initialize(self): """Initialize the FSDP model, then apply XPU-specific workarounds. - The key workaround: force sum-based gradient reduction. - This is needed because xccl does not support ReduceOp.AVG. - The FSDP wrapper will use SUM + manual division instead. + The key workaround: force sum-based gradient reduction, root module + and every per-layer FSDP2 wrapper. This is needed because xccl does + not support ReduceOp.AVG. The FSDP wrapper will use SUM + manual + division instead. """ super().initialize() # xccl does not support ReduceOp.AVG; force sum-based reduction - if hasattr(self.model, "set_force_sum_reduction_for_comms"): - self.model.set_force_sum_reduction_for_comms(True) - logger.info("Enabled force_sum_reduction_for_comms for XPU") + _force_sum_reduction_on_all_fsdp_modules(self.module) @EngineRegistry.register(model_type="value_model", backend=["fsdp", "fsdp2"], device="xpu", vendor="intel") @@ -91,5 +127,4 @@ def __init__( def initialize(self): """Initialize the FSDP value model, then apply xccl workaround.""" super().initialize() - if hasattr(self.model, "set_force_sum_reduction_for_comms"): - self.model.set_force_sum_reduction_for_comms(True) + _force_sum_reduction_on_all_fsdp_modules(self.module) diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py new file mode 100644 index 00000000..3d64fd87 --- /dev/null +++ b/verl_hardware_plugin/patches/__init__.py @@ -0,0 +1,40 @@ +# Copyright (c) 2026 BAAI. All rights reserved. +# Licensed under the Apache License, Version 2.0. + +"""Plugin-side monkeypatches for Intel XPU reduce_avg + attention support, +applied with zero changes to verl-core and with no dependency on any new +PlatformBase hook. + +This is the "how far can pure monkeypatch actually go" counterpart to the +PlatformBase-hook approach used elsewhere in this plugin +(is_reduce_avg_supported/attention_utils_module, gated behind the +still-unmerged verl-project/verl#7917). See +docs/design/xpu-monkeypatch-reduce-avg-attention.md for the per-capability +verdict. + +The VTune/dist_profiler_cls monkeypatch is intentionally not part of this +package — see docs/design/xpu-monkeypatch-vtune-profiler.md, developed as a +separate branch/PR so the two can be reviewed and merged independently. + +Each patch module is independently guarded by an XPU-availability check, and +apply_all() is idempotent, so importing this package is safe even when +another vendor's plugin (or no accelerator at all) is active in the same +process. +""" + +import logging +import os + +logger = logging.getLogger(__name__) +logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN")) + + +def apply_all() -> None: + """Apply every XPU monkeypatch. Safe to call multiple times or on non-XPU hosts.""" + from . import attention_patch_xpu, reduce_avg_allreduce_patch_xpu + + for module in (attention_patch_xpu, reduce_avg_allreduce_patch_xpu): + try: + module.apply() + except Exception as e: + logger.debug("%s.apply() skipped: %s", module.__name__, e) diff --git a/verl_hardware_plugin/patches/attention_patch_xpu.py b/verl_hardware_plugin/patches/attention_patch_xpu.py new file mode 100644 index 00000000..38dda758 --- /dev/null +++ b/verl_hardware_plugin/patches/attention_patch_xpu.py @@ -0,0 +1,55 @@ +# Copyright (c) 2026 BAAI. All rights reserved. +# Licensed under the Apache License, Version 2.0. + +"""Monkeypatch to route Intel XPU through the NPU flash-attention shim for +rmpad attention padding, with zero verl-core changes. + +Dispatch shape that makes this safe (verl/utils/attention_utils.py): + verl.utils.attention_utils.index_first_axis/pad_input/rearrange/unpad_input + are stable public wrappers. Each one calls the private + `_get_attention_functions()` fresh on *every* invocation -- it is not + cached at import time. That means patching `_get_attention_functions` + itself takes effect for every caller regardless of whether that caller + did `from verl.utils.attention_utils import pad_input` before or after + this patch runs. This is the one hook where import order genuinely does + not matter. +""" + +import logging +import os + +logger = logging.getLogger(__name__) +logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN")) + +_applied = False + + +def _xpu_available() -> bool: + import torch + + return hasattr(torch, "xpu") and torch.xpu.is_available() + + +def apply() -> None: + global _applied + if _applied: + return + + import verl.utils.attention_utils as attention_utils + + original_get_attention_functions = attention_utils._get_attention_functions + + def _patched_get_attention_functions(): + if _xpu_available(): + from verl.utils.npu_flash_attn_utils import index_first_axis, pad_input, rearrange, unpad_input + + attention_utils._index_first_axis = index_first_axis + attention_utils._pad_input = pad_input + attention_utils._rearrange = rearrange + attention_utils._unpad_input = unpad_input + return index_first_axis, pad_input, rearrange, unpad_input + return original_get_attention_functions() + + attention_utils._get_attention_functions = _patched_get_attention_functions + _applied = True + logger.info("[verl_hardware_plugin] Patched attention_utils._get_attention_functions for XPU") diff --git a/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py b/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py new file mode 100644 index 00000000..c7ec8f73 --- /dev/null +++ b/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py @@ -0,0 +1,72 @@ +# Copyright (c) 2026 BAAI. All rights reserved. +# Licensed under the Apache License, Version 2.0. + +"""Monkeypatch torch.distributed.all_reduce(op=AVG) -> SUM + manual divide, +process-wide, for Intel XPU (xccl does not implement ReduceOp.AVG). + +This is the one patch in this package worth calling "doable, but with a real +design smell" rather than "clean": + +- verl/workers/engine_workers.py, verl/trainer/sft_trainer.py, and + verl/utils/profiler/performance.py::reduce_timing each call + `torch.distributed.all_reduce(x, op=ReduceOp.AVG, ...)` as a literal inline + expression -- there is no per-site wrapper or dispatch object to patch, the + way there is for attention padding. +- Patching `torch.distributed.all_reduce` itself is robust to import order + (every call site does `dist.all_reduce(...)`, an attribute lookup on the + `torch.distributed` module performed at call time, not at each file's own + import time), so this does take effect regardless of load order. +- The cost is real: this changes the process-wide meaning of + `ReduceOp.AVG` for every call site in the process, XPU-related or not. + Anyone reading verl-core's source sees real `ReduceOp.AVG` semantics and + silently gets SUM+divide instead. That is a legitimate argument for fixing + this in verl-core instead (verl-project/verl#7917's + `is_reduce_avg_supported()` hook covers exactly these 3 call sites) -- this + patch exists to show it is *possible* without core changes, not to claim + it is the better design. + +async_op=True is intentionally unsupported: correctly dividing the result +requires the collective to have already completed, which async_op explicitly +defers. Call sites using async_op=True with op=AVG fall through to the +original (broken-on-xccl) behavior rather than silently producing a wrong +answer with no error. +""" + +import logging +import os + +logger = logging.getLogger(__name__) +logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN")) + +_applied = False + + +def _xpu_available() -> bool: + import torch + + return hasattr(torch, "xpu") and torch.xpu.is_available() + + +def apply() -> None: + global _applied + if _applied: + return + if not _xpu_available(): + return + + import torch.distributed as dist + + original_all_reduce = dist.all_reduce + + def _patched_all_reduce(tensor, op=dist.ReduceOp.SUM, group=None, async_op=False): + if op != dist.ReduceOp.AVG or async_op: + return original_all_reduce(tensor, op=op, group=group, async_op=async_op) + + world_size = dist.get_world_size(group=group) + result = original_all_reduce(tensor, op=dist.ReduceOp.SUM, group=group, async_op=False) + tensor.div_(world_size) + return result + + dist.all_reduce = _patched_all_reduce + _applied = True + logger.info("[verl_hardware_plugin] Patched torch.distributed.all_reduce(op=AVG) for XPU/xccl") From de6f66ac4b6c66d83bb682635174a42f5cea6a53 Mon Sep 17 00:00:00 2001 From: kahlun Date: Thu, 24 Sep 2026 03:20:31 +0000 Subject: [PATCH 02/11] refactor: narrow this branch to reduce_avg only, drop the attention patch attention_utils_module() is staying a PlatformBase hook (see PR #22, and verl-project/verl#7917's updated description) -- it passed the maintainer's bar on verl-hardware-plugin#26 for a genuinely hardware-agnostic capability that couldn't be done cleanly from the plugin alone. Carrying a monkeypatch alternative for it here too just duplicates an already-accepted, working solution with no benefit. is_reduce_avg_supported() didn't pass that bar and is this branch's actual job: the xccl AVG gap can be fully handled from the plugin (FSDP2 via an engine-level method override, the 3 remaining raw dist.all_reduce(op=AVG) sites via a scoped, documented monkeypatch). Renamed the design doc to match (xpu-monkeypatch-reduce-avg.md) and folded in a note that this branch is now the sole owner of the fsdp_xpu.py self.model->self.module fix -- it's been dropped from #22's smaller, root-only version of the same fix now that #22 no longer touches this file for any other reason. --- ...ntion.md => xpu-monkeypatch-reduce-avg.md} | 0 verl_hardware_plugin/patches/__init__.py | 32 +++++------ .../patches/attention_patch_xpu.py | 55 ------------------- 3 files changed, 16 insertions(+), 71 deletions(-) rename docs/design/{xpu-monkeypatch-reduce-avg-attention.md => xpu-monkeypatch-reduce-avg.md} (100%) delete mode 100644 verl_hardware_plugin/patches/attention_patch_xpu.py diff --git a/docs/design/xpu-monkeypatch-reduce-avg-attention.md b/docs/design/xpu-monkeypatch-reduce-avg.md similarity index 100% rename from docs/design/xpu-monkeypatch-reduce-avg-attention.md rename to docs/design/xpu-monkeypatch-reduce-avg.md diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py index 3d64fd87..8c8f52f9 100644 --- a/verl_hardware_plugin/patches/__init__.py +++ b/verl_hardware_plugin/patches/__init__.py @@ -1,25 +1,25 @@ # Copyright (c) 2026 BAAI. All rights reserved. # Licensed under the Apache License, Version 2.0. -"""Plugin-side monkeypatches for Intel XPU reduce_avg + attention support, -applied with zero changes to verl-core and with no dependency on any new -PlatformBase hook. - -This is the "how far can pure monkeypatch actually go" counterpart to the -PlatformBase-hook approach used elsewhere in this plugin -(is_reduce_avg_supported/attention_utils_module, gated behind the -still-unmerged verl-project/verl#7917). See -docs/design/xpu-monkeypatch-reduce-avg-attention.md for the per-capability -verdict. +"""Plugin-side monkeypatch for Intel XPU reduce_avg support, applied with +zero changes to verl-core and with no dependency on any new PlatformBase +hook. + +Per the maintainer's ruling on verl-hardware-plugin#26 (2026-09-23): default +new hardware backends to plugin-side monkeypatching, reserving new +PlatformBase hooks for capabilities that genuinely can't be done from the +plugin. `attention_utils_module()` passed that bar and stays a hook +(implemented in `platforms/platform_xpu.py`, see PR #22) — this package only +covers `is_reduce_avg_supported`, the one capability that didn't. See +docs/design/xpu-monkeypatch-reduce-avg.md for the full reasoning. The VTune/dist_profiler_cls monkeypatch is intentionally not part of this package — see docs/design/xpu-monkeypatch-vtune-profiler.md, developed as a separate branch/PR so the two can be reviewed and merged independently. -Each patch module is independently guarded by an XPU-availability check, and -apply_all() is idempotent, so importing this package is safe even when -another vendor's plugin (or no accelerator at all) is active in the same -process. +The patch module is guarded by an XPU-availability check, and apply_all() is +idempotent, so importing this package is safe even when another vendor's +plugin (or no accelerator at all) is active in the same process. """ import logging @@ -31,9 +31,9 @@ def apply_all() -> None: """Apply every XPU monkeypatch. Safe to call multiple times or on non-XPU hosts.""" - from . import attention_patch_xpu, reduce_avg_allreduce_patch_xpu + from . import reduce_avg_allreduce_patch_xpu - for module in (attention_patch_xpu, reduce_avg_allreduce_patch_xpu): + for module in (reduce_avg_allreduce_patch_xpu,): try: module.apply() except Exception as e: diff --git a/verl_hardware_plugin/patches/attention_patch_xpu.py b/verl_hardware_plugin/patches/attention_patch_xpu.py deleted file mode 100644 index 38dda758..00000000 --- a/verl_hardware_plugin/patches/attention_patch_xpu.py +++ /dev/null @@ -1,55 +0,0 @@ -# Copyright (c) 2026 BAAI. All rights reserved. -# Licensed under the Apache License, Version 2.0. - -"""Monkeypatch to route Intel XPU through the NPU flash-attention shim for -rmpad attention padding, with zero verl-core changes. - -Dispatch shape that makes this safe (verl/utils/attention_utils.py): - verl.utils.attention_utils.index_first_axis/pad_input/rearrange/unpad_input - are stable public wrappers. Each one calls the private - `_get_attention_functions()` fresh on *every* invocation -- it is not - cached at import time. That means patching `_get_attention_functions` - itself takes effect for every caller regardless of whether that caller - did `from verl.utils.attention_utils import pad_input` before or after - this patch runs. This is the one hook where import order genuinely does - not matter. -""" - -import logging -import os - -logger = logging.getLogger(__name__) -logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN")) - -_applied = False - - -def _xpu_available() -> bool: - import torch - - return hasattr(torch, "xpu") and torch.xpu.is_available() - - -def apply() -> None: - global _applied - if _applied: - return - - import verl.utils.attention_utils as attention_utils - - original_get_attention_functions = attention_utils._get_attention_functions - - def _patched_get_attention_functions(): - if _xpu_available(): - from verl.utils.npu_flash_attn_utils import index_first_axis, pad_input, rearrange, unpad_input - - attention_utils._index_first_axis = index_first_axis - attention_utils._pad_input = pad_input - attention_utils._rearrange = rearrange - attention_utils._unpad_input = unpad_input - return index_first_axis, pad_input, rearrange, unpad_input - return original_get_attention_functions() - - attention_utils._get_attention_functions = _patched_get_attention_functions - _applied = True - logger.info("[verl_hardware_plugin] Patched attention_utils._get_attention_functions for XPU") From ff00a015ab81c5982524de53b0ad62292ac07332 Mon Sep 17 00:00:00 2001 From: kahlun Date: Thu, 24 Sep 2026 03:42:40 +0000 Subject: [PATCH 03/11] fix(xpu): apply the reduce_avg patch on platform selection, not import verl_hardware_plugin/__init__.py is shared across every vendor in this plugin, so calling apply_all_patches() there fired the patch on any host where torch.xpu merely happened to be importable -- regardless of which platform verl actually selected for the run. A mixed host with e.g. VERL_PLATFORM=nvidia set explicitly would still get its process-wide torch.distributed.all_reduce(op=AVG) semantics changed for no reason, compounding the design smell this patch already carries. verl.plugin.platform.platform_manager._create_platform() only constructs PlatformXPU() once platform detection has actually picked "intel" for this process, so moving the call into PlatformXPU.__init__ ties it to real platform selection instead. The narrow remaining edge case -- auto-detection probing PlatformXPU() to check availability, then falling through to a different platform -- is self-mitigating: the patch's own _xpu_available() gate fails in the same underlying probe that made is_platform_available() fail. Also: - Add unit tests for both the new wiring (PlatformXPU() triggers the patch; the plugin's top-level __init__ no longer binds it directly) and the reduce_avg_allreduce_patch_xpu wrapper itself (AVG -> SUM+divide, other ops pass through unchanged, async_op=True falls through unpatched, no-op when XPU unavailable, idempotent). 75 passed. - Remove docs/design/xpu-monkeypatch-reduce-avg.md. This repo has no precedent for keeping RFC/experiment-verdict docs in the merged tree (confirmed: no docs/design/ directory exists on main, in any prior PR). Its reasoning already lives in this PR's description. --- docs/design/xpu-monkeypatch-reduce-avg.md | 80 ------------ tests/test_plugin_registration.py | 122 ++++++++++++++++++ verl_hardware_plugin/__init__.py | 2 - verl_hardware_plugin/patches/__init__.py | 20 ++- .../platforms/platform_xpu.py | 16 +++ 5 files changed, 151 insertions(+), 89 deletions(-) delete mode 100644 docs/design/xpu-monkeypatch-reduce-avg.md diff --git a/docs/design/xpu-monkeypatch-reduce-avg.md b/docs/design/xpu-monkeypatch-reduce-avg.md deleted file mode 100644 index f5a81a6d..00000000 --- a/docs/design/xpu-monkeypatch-reduce-avg.md +++ /dev/null @@ -1,80 +0,0 @@ -# Experiment: reduce_avg + attention support for XPU, pure monkeypatch - -## Context - -`verl-hardware-plugin#26` asked maintainers whether Intel XPU should extend -`PlatformBase` in `verl-core` (the TPU precedent) or stay entirely inside -this plugin via monkeypatch (the MLU precedent). The maintainer's answer -(2026-09-23): default to the plugin, keep `verl-project/verl#7917` (which -adds `is_reduce_avg_supported`/`attention_utils_module`/`profiler_markers`/ -`dist_profiler_cls` to `PlatformBase`) narrowed to only what truly can't be -done from here. - -This branch covers two of those four capabilities — -`is_reduce_avg_supported` and `attention_utils_module` — split out from the -original combined experiment (`experiment/xpu-monkeypatch-no-core-hooks`) -so this half, which has no unresolved design smell beyond the one flagged -below, can be reviewed and hardware-validated independently of the VTune -profiler half (`docs/design/xpu-monkeypatch-vtune-profiler.md`), which has -an open blocker. - -Does not touch `feature/xpu-vtune-avg-attention` (the branch behind open -PR #22, which uses the hook-based approach and depends on #7917 merging). -Built from `main` so it can be compared directly for footprint and -core-change dependency. - -## Verdict per capability - -| Capability | Mechanism | Notes | -|---|---|---| -| `attention_utils_module` | Patch `verl.utils.attention_utils._get_attention_functions` | Clean. That function is re-invoked on every call by the public wrappers, so patch order doesn't matter. | -| `is_reduce_avg_supported` (FSDP2) | Not a patch at all — extended `verl_hardware_plugin/engines/fsdp_xpu.py`'s existing `initialize()` to loop over every FSDP2-wrapped submodule, not just the root | See below — this is *better* than the monkeypatch this branch originally planned, and matches the per-layer+root fix already validated on real B60 hardware via a 12-run controlled A/B (6/6 pass with the fix, 6/6 fail without, on `run_grpo_fsdp2_intel_gpu.sh`). | -| `is_reduce_avg_supported` (3 raw `dist.all_reduce(op=AVG)` sites) | Patch `torch.distributed.all_reduce` process-wide when `op=AVG` | Robust to import order (attribute lookup at call time), but changes global `ReduceOp.AVG` semantics for the whole process. Real design smell, documented in `patches/reduce_avg_allreduce_patch_xpu.py`. This is the one piece here still worth a narrow core-hook ask if the process-wide semantics change is judged unacceptable on review. | - -Net: both capabilities work today with zero `verl-core` changes. - -## Why the FSDP2 fix isn't a patch - -The first draft of this plan (matching the original proposal doc) was to -wrap `verl.utils.fsdp_utils.apply_fsdp2`: call the original, then walk -`model.modules()` and call `set_force_sum_reduction_for_comms(True)`. - -That doesn't actually work reliably: `verl/workers/engine/fsdp/ -transformer_impl.py` does `from verl.utils.fsdp_utils import apply_fsdp2` at -its own import time. Reassigning `verl.utils.fsdp_utils.apply_fsdp2` after -that has already happened has no effect on the name bound inside -`transformer_impl.py`. - -This plugin doesn't need that patch at all, though: `fsdp_xpu.py`'s engine -subclasses already call `set_force_sum_reduction_for_comms` on the -constructed model *after* `super().initialize()` returns — from inside their -own `EngineRegistry`-registered class, which verl-core calls directly (no -patching involved). This branch extends that existing, already-real -mechanism to loop over every FSDP2-wrapped submodule instead of only the -root (`_force_sum_reduction_on_all_fsdp_modules` in `fsdp_xpu.py`), achieved -here with no patch and no hook. - -Note: PR #22's current `fsdp_xpu.py` (the hook-based branch) only fixes a -`self.model`→`self.module` attribute bug and still calls -`set_force_sum_reduction_for_comms` on the root only — it does not yet -include this per-layer loop. Whichever branch ships, this fix should land; -it isn't specific to the monkeypatch-vs-hook question. - -## Comparison - -| | Files touched outside this plugin | Depends on #7917 merging | -|---|---|---| -| PR #22 (`feature/xpu-vtune-avg-attention`) | 0 (but the `is_reduce_avg_supported`/`attention_utils_module` hooks are dead code without #7917) | Yes | -| This branch | 0 | No | - -## Not yet done - -- No hardware validation of the `attention_utils_module` patch or the raw - `dist.all_reduce(AVG)` patch — reasoned through against `verl-core`'s real - `main` source, not run on a B60. The FSDP2 per-layer fix itself is already - hardware-validated (see the 12-run A/B cited above); only its - monkeypatch-vs-engine-override framing here is new. -- No unit tests yet for `patches/`. -- Whether to still ask for a narrow core hook covering the 3 raw - `dist.all_reduce(AVG)` call sites, given the process-wide semantics change - this patch makes, is an open question for review. diff --git a/tests/test_plugin_registration.py b/tests/test_plugin_registration.py index b2288b9d..34f9dfa5 100644 --- a/tests/test_plugin_registration.py +++ b/tests/test_plugin_registration.py @@ -689,5 +689,127 @@ def test_raises_on_whitelist_and_blacklist(self): may_enable_flag_gems(phase="training") +class TestReduceAvgPatchWiring: + """The reduce_avg monkeypatch must fire on platform *selection*, not on + mere XPU hardware/SDK *presence*. + + verl.plugin.platform.platform_manager._create_platform() only constructs + PlatformXPU() once platform detection (VERL_PLATFORM=intel, or + auto-detection) has actually picked "intel" for this process. Applying + the patch from the shared verl_hardware_plugin/__init__.py instead would + fire on any host where XPU merely happens to be importable, even when a + different platform is selected for the run -- e.g. VERL_PLATFORM=nvidia + explicitly set on a mixed host would still get its process-wide + torch.distributed.all_reduce(op=AVG) semantics changed for no reason. + """ + + def test_platform_xpu_init_applies_patches(self): + from verl_hardware_plugin.platforms.platform_xpu import PlatformXPU + + with mock.patch("verl_hardware_plugin.patches.apply_all") as fake_apply_all: + PlatformXPU() + + fake_apply_all.assert_called_once() + + def test_plugin_init_does_not_bind_apply_all_patches(self): + import verl_hardware_plugin + + assert not hasattr(verl_hardware_plugin, "apply_all_patches") + + +class TestReduceAvgAllReducePatch: + """torch.distributed.all_reduce(op=AVG) -> SUM + manual divide, XPU only. + + _applied is process-global module state (the same property that makes + this patch risky in production -- see the module's own docstring), so + each test must save/restore torch.distributed.all_reduce and reset the + flag, or tests would leak into each other. + """ + + @pytest.fixture(autouse=True) + def _reset_patch_state(self): + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + original_all_reduce = dist.all_reduce + patch_mod._applied = False + yield + dist.all_reduce = original_all_reduce + patch_mod._applied = False + + def test_noop_when_xpu_unavailable(self): + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + before = dist.all_reduce + with mock.patch.object(patch_mod, "_xpu_available", return_value=False): + patch_mod.apply() + + assert dist.all_reduce is before + + def test_avg_becomes_sum_plus_divide(self): + import torch + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + fake_original = mock.MagicMock() + tensor = torch.tensor([4.0]) + with mock.patch.object(patch_mod, "_xpu_available", return_value=True): + with mock.patch.object(dist, "all_reduce", fake_original): + patch_mod.apply() + patched = dist.all_reduce + assert patched is not fake_original + + with mock.patch.object(dist, "get_world_size", return_value=4): + patched(tensor, op=dist.ReduceOp.AVG) + + fake_original.assert_called_once_with(tensor, op=dist.ReduceOp.SUM, group=None, async_op=False) + assert tensor.item() == 1.0 + + def test_non_avg_op_passes_through_unchanged(self): + import torch + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + fake_original = mock.MagicMock() + tensor = torch.tensor([4.0]) + with mock.patch.object(patch_mod, "_xpu_available", return_value=True): + with mock.patch.object(dist, "all_reduce", fake_original): + patch_mod.apply() + dist.all_reduce(tensor, op=dist.ReduceOp.SUM, group="g") + + fake_original.assert_called_once_with(tensor, op=dist.ReduceOp.SUM, group="g", async_op=False) + + def test_async_avg_falls_through_unpatched(self): + import torch + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + fake_original = mock.MagicMock() + tensor = torch.tensor([4.0]) + with mock.patch.object(patch_mod, "_xpu_available", return_value=True): + with mock.patch.object(dist, "all_reduce", fake_original): + patch_mod.apply() + dist.all_reduce(tensor, op=dist.ReduceOp.AVG, async_op=True) + + fake_original.assert_called_once_with(tensor, op=dist.ReduceOp.AVG, group=None, async_op=True) + + def test_idempotent(self): + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + with mock.patch.object(patch_mod, "_xpu_available", return_value=True): + patch_mod.apply() + patched_once = dist.all_reduce + patch_mod.apply() + assert dist.all_reduce is patched_once + + if __name__ == "__main__": pytest.main([__file__, "-v"]) diff --git a/verl_hardware_plugin/__init__.py b/verl_hardware_plugin/__init__.py index 90b7e4ee..90ee349b 100644 --- a/verl_hardware_plugin/__init__.py +++ b/verl_hardware_plugin/__init__.py @@ -14,7 +14,6 @@ import os from verl_hardware_plugin.engines import register_all_engines -from verl_hardware_plugin.patches import apply_all as apply_all_patches from verl_hardware_plugin.platforms import register_all_platforms from verl_hardware_plugin.profilers import register_all_profiles @@ -24,6 +23,5 @@ register_all_platforms() register_all_engines() register_all_profiles() -apply_all_patches() logger.info("verl-hardware-plugin loaded successfully") diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py index 8c8f52f9..4a2575eb 100644 --- a/verl_hardware_plugin/patches/__init__.py +++ b/verl_hardware_plugin/patches/__init__.py @@ -10,16 +10,22 @@ PlatformBase hooks for capabilities that genuinely can't be done from the plugin. `attention_utils_module()` passed that bar and stays a hook (implemented in `platforms/platform_xpu.py`, see PR #22) — this package only -covers `is_reduce_avg_supported`, the one capability that didn't. See -docs/design/xpu-monkeypatch-reduce-avg.md for the full reasoning. +covers `is_reduce_avg_supported`, the one capability that didn't. -The VTune/dist_profiler_cls monkeypatch is intentionally not part of this -package — see docs/design/xpu-monkeypatch-vtune-profiler.md, developed as a -separate branch/PR so the two can be reviewed and merged independently. +Call `apply_all()` from `PlatformXPU.__init__`, not from this plugin's +top-level `verl_hardware_plugin/__init__.py`: the latter runs on any host +where XPU merely happens to be importable, regardless of which platform +verl actually selects for the run, whereas `PlatformXPU()` is only +constructed once XPU is the selected platform (see +`verl.plugin.platform.platform_manager._create_platform`). + +The VTune/dist_profiler_cls monkeypatch is a separate, unrelated concern and +intentionally not part of this package — it ships as its own branch/PR so +the two can be reviewed and merged independently. The patch module is guarded by an XPU-availability check, and apply_all() is -idempotent, so importing this package is safe even when another vendor's -plugin (or no accelerator at all) is active in the same process. +idempotent, so calling it is safe even when another vendor's platform (or no +accelerator at all) ends up selected in the same process. """ import logging diff --git a/verl_hardware_plugin/platforms/platform_xpu.py b/verl_hardware_plugin/platforms/platform_xpu.py index 1f44e620..88eb7f5b 100644 --- a/verl_hardware_plugin/platforms/platform_xpu.py +++ b/verl_hardware_plugin/platforms/platform_xpu.py @@ -77,6 +77,22 @@ class PlatformXPU(PlatformBase): force_sum_reduction to work around this limitation. """ + def __init__(self) -> None: + # Apply plugin-side monkeypatches only now, when XPU is actually the + # platform verl selected for this process (verl.plugin.platform. + # platform_manager._create_platform() only constructs PlatformXPU() + # when platform detection -- explicit VERL_PLATFORM=intel or + # auto-detection -- picked "intel"). Applying this from the shared + # verl_hardware_plugin/__init__.py instead would fire on any host + # where XPU hardware/SDK merely happens to be present, even if a + # different platform ends up selected for this run -- e.g. a mixed + # host with VERL_PLATFORM=nvidia explicitly set would still get its + # process-wide torch.distributed.all_reduce(op=AVG) semantics + # changed for no reason. See patches/reduce_avg_allreduce_patch_xpu.py. + from verl_hardware_plugin.patches import apply_all as apply_all_patches + + apply_all_patches() + # ------------------------------------------------------------------ # Core device management # ------------------------------------------------------------------ From a7669ab8f09cba7ba8172c7bc4a0af6bb910b1a4 Mon Sep 17 00:00:00 2001 From: kahlun Date: Thu, 24 Sep 2026 03:54:03 +0000 Subject: [PATCH 04/11] refactor: drop the patches.apply_all() indirection, call the module directly apply_all() had exactly one caller (PlatformXPU.__init__) and exactly one patch module (reduce_avg_allreduce_patch_xpu) -- no other vendor in this plugin has any monkeypatches. The generic "apply every patch" name and loop implied a cross-vendor aggregator like register_all_platforms()/ register_all_engines(), which genuinely do need to register every vendor up front since verl-core's registries look classes up by name after the fact. Patches don't share that constraint: they only ever matter to the one platform that already got selected, so each platform's own __init__ calling its own patch module directly is both simpler and more honest about what's actually happening than routing through a same-package wrapper that only ever had one real implementation behind it. verl_hardware_plugin/patches/__init__.py is now a plain package docstring pointing at reduce_avg_allreduce_patch_xpu, with no code. Updated the wiring test to mock the real target (reduce_avg_allreduce_patch_xpu.apply) instead of the removed wrapper. 75 passed. --- tests/test_plugin_registration.py | 9 ++- verl_hardware_plugin/patches/__init__.py | 58 +++++-------------- .../platforms/platform_xpu.py | 12 ++-- 3 files changed, 28 insertions(+), 51 deletions(-) diff --git a/tests/test_plugin_registration.py b/tests/test_plugin_registration.py index 34f9dfa5..2b55e151 100644 --- a/tests/test_plugin_registration.py +++ b/tests/test_plugin_registration.py @@ -706,15 +706,18 @@ class TestReduceAvgPatchWiring: def test_platform_xpu_init_applies_patches(self): from verl_hardware_plugin.platforms.platform_xpu import PlatformXPU - with mock.patch("verl_hardware_plugin.patches.apply_all") as fake_apply_all: + with mock.patch( + "verl_hardware_plugin.patches.reduce_avg_allreduce_patch_xpu.apply" + ) as fake_apply: PlatformXPU() - fake_apply_all.assert_called_once() + fake_apply.assert_called_once() - def test_plugin_init_does_not_bind_apply_all_patches(self): + def test_plugin_init_does_not_bind_reduce_avg_patch(self): import verl_hardware_plugin assert not hasattr(verl_hardware_plugin, "apply_all_patches") + assert not hasattr(verl_hardware_plugin, "reduce_avg_allreduce_patch_xpu") class TestReduceAvgAllReducePatch: diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py index 4a2575eb..6f9d1d8c 100644 --- a/verl_hardware_plugin/patches/__init__.py +++ b/verl_hardware_plugin/patches/__init__.py @@ -1,46 +1,20 @@ # Copyright (c) 2026 BAAI. All rights reserved. # Licensed under the Apache License, Version 2.0. -"""Plugin-side monkeypatch for Intel XPU reduce_avg support, applied with -zero changes to verl-core and with no dependency on any new PlatformBase -hook. - -Per the maintainer's ruling on verl-hardware-plugin#26 (2026-09-23): default -new hardware backends to plugin-side monkeypatching, reserving new -PlatformBase hooks for capabilities that genuinely can't be done from the -plugin. `attention_utils_module()` passed that bar and stays a hook -(implemented in `platforms/platform_xpu.py`, see PR #22) — this package only -covers `is_reduce_avg_supported`, the one capability that didn't. - -Call `apply_all()` from `PlatformXPU.__init__`, not from this plugin's -top-level `verl_hardware_plugin/__init__.py`: the latter runs on any host -where XPU merely happens to be importable, regardless of which platform -verl actually selects for the run, whereas `PlatformXPU()` is only -constructed once XPU is the selected platform (see -`verl.plugin.platform.platform_manager._create_platform`). - -The VTune/dist_profiler_cls monkeypatch is a separate, unrelated concern and -intentionally not part of this package — it ships as its own branch/PR so -the two can be reviewed and merged independently. - -The patch module is guarded by an XPU-availability check, and apply_all() is -idempotent, so calling it is safe even when another vendor's platform (or no -accelerator at all) ends up selected in the same process. +"""Plugin-side monkeypatches, one module per capability, applied only from +the platform class that needs them -- not from this package's `__init__.py` +and not from this plugin's shared top-level `verl_hardware_plugin/__init__.py`. + +`PlatformXXX()` is only constructed once `verl.plugin.platform. +platform_manager._create_platform()` has actually selected that platform for +the process (see `PlatformXPU.__init__` in `platforms/platform_xpu.py`). +Applying a patch from a shared location instead would fire on any host where +the corresponding hardware/SDK merely happens to be importable, regardless of +which platform verl actually selects for the run. + +Currently contains one module: + +- `reduce_avg_allreduce_patch_xpu`: Intel XPU's `is_reduce_avg_supported` + replacement. See its own docstring for what it does and why it's a + monkeypatch rather than a `PlatformBase` hook (verl-hardware-plugin#26). """ - -import logging -import os - -logger = logging.getLogger(__name__) -logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN")) - - -def apply_all() -> None: - """Apply every XPU monkeypatch. Safe to call multiple times or on non-XPU hosts.""" - from . import reduce_avg_allreduce_patch_xpu - - for module in (reduce_avg_allreduce_patch_xpu,): - try: - module.apply() - except Exception as e: - logger.debug("%s.apply() skipped: %s", module.__name__, e) diff --git a/verl_hardware_plugin/platforms/platform_xpu.py b/verl_hardware_plugin/platforms/platform_xpu.py index 88eb7f5b..3c095865 100644 --- a/verl_hardware_plugin/platforms/platform_xpu.py +++ b/verl_hardware_plugin/platforms/platform_xpu.py @@ -78,20 +78,20 @@ class PlatformXPU(PlatformBase): """ def __init__(self) -> None: - # Apply plugin-side monkeypatches only now, when XPU is actually the - # platform verl selected for this process (verl.plugin.platform. + # Apply this platform's monkeypatch only now, when XPU is actually + # the platform verl selected for this process (verl.plugin.platform. # platform_manager._create_platform() only constructs PlatformXPU() # when platform detection -- explicit VERL_PLATFORM=intel or - # auto-detection -- picked "intel"). Applying this from the shared + # auto-detection -- picked "intel"). Applying it from the shared # verl_hardware_plugin/__init__.py instead would fire on any host # where XPU hardware/SDK merely happens to be present, even if a # different platform ends up selected for this run -- e.g. a mixed # host with VERL_PLATFORM=nvidia explicitly set would still get its # process-wide torch.distributed.all_reduce(op=AVG) semantics - # changed for no reason. See patches/reduce_avg_allreduce_patch_xpu.py. - from verl_hardware_plugin.patches import apply_all as apply_all_patches + # changed for no reason. + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu - apply_all_patches() + reduce_avg_allreduce_patch_xpu.apply() # ------------------------------------------------------------------ # Core device management From 08fb6a92e3aa486baa2a8409466983955ed481d6 Mon Sep 17 00:00:00 2001 From: kahlun Date: Thu, 24 Sep 2026 04:07:00 +0000 Subject: [PATCH 05/11] fix(xpu): scope the reduce_avg patch to xccl process groups only _patched_all_reduce only checked op != AVG and async_op -- it never checked which backend the target group actually uses. Once applied, it silently intercepted op=AVG for every process group in the process, not just xccl ones: a gloo group in the same process (e.g. a CPU-only Ray actor's coordination group alongside a GPU worker's xccl group) would get routed through SUM+divide too, even though gloo/nccl handle AVG natively and never needed the workaround. Added dist.get_backend(group) != "xccl" to the short-circuit condition so non-xccl groups fall through to the original, untouched. Short-circuit evaluation means only the actual substitution path needs get_backend mocked in tests -- the passthrough tests (non-AVG op, async_op=True) never reach it, since op != AVG or async_op already short-circuits first. Added a test for a gloo-backed group with op=AVG to confirm it now passes through unpatched. 76 passed. --- tests/test_plugin_registration.py | 26 +++++++++++++++--- .../patches/reduce_avg_allreduce_patch_xpu.py | 27 +++++++++++-------- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/tests/test_plugin_registration.py b/tests/test_plugin_registration.py index 2b55e151..f36f875d 100644 --- a/tests/test_plugin_registration.py +++ b/tests/test_plugin_registration.py @@ -721,7 +721,9 @@ def test_plugin_init_does_not_bind_reduce_avg_patch(self): class TestReduceAvgAllReducePatch: - """torch.distributed.all_reduce(op=AVG) -> SUM + manual divide, XPU only. + """torch.distributed.all_reduce(op=AVG) -> SUM + manual divide, scoped to + xccl process groups only -- a gloo/nccl group in the same process (e.g. a + CPU-only Ray actor's coordination group) must keep native AVG. _applied is process-global module state (the same property that makes this patch risky in production -- see the module's own docstring), so @@ -766,12 +768,30 @@ def test_avg_becomes_sum_plus_divide(self): patched = dist.all_reduce assert patched is not fake_original - with mock.patch.object(dist, "get_world_size", return_value=4): - patched(tensor, op=dist.ReduceOp.AVG) + with mock.patch.object(dist, "get_backend", return_value="xccl"): + with mock.patch.object(dist, "get_world_size", return_value=4): + patched(tensor, op=dist.ReduceOp.AVG) fake_original.assert_called_once_with(tensor, op=dist.ReduceOp.SUM, group=None, async_op=False) assert tensor.item() == 1.0 + def test_non_xccl_backend_passes_through_even_with_avg(self): + """A gloo/nccl group in the same process must keep native AVG untouched.""" + import torch + import torch.distributed as dist + + from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + + fake_original = mock.MagicMock() + tensor = torch.tensor([4.0]) + with mock.patch.object(patch_mod, "_xpu_available", return_value=True): + with mock.patch.object(dist, "all_reduce", fake_original): + patch_mod.apply() + with mock.patch.object(dist, "get_backend", return_value="gloo"): + dist.all_reduce(tensor, op=dist.ReduceOp.AVG, group="cpu_group") + + fake_original.assert_called_once_with(tensor, op=dist.ReduceOp.AVG, group="cpu_group", async_op=False) + def test_non_avg_op_passes_through_unchanged(self): import torch import torch.distributed as dist diff --git a/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py b/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py index c7ec8f73..3e50c7c1 100644 --- a/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py +++ b/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py @@ -2,7 +2,8 @@ # Licensed under the Apache License, Version 2.0. """Monkeypatch torch.distributed.all_reduce(op=AVG) -> SUM + manual divide, -process-wide, for Intel XPU (xccl does not implement ReduceOp.AVG). +scoped to xccl process groups, for Intel XPU (xccl does not implement +ReduceOp.AVG). This is the one patch in this package worth calling "doable, but with a real design smell" rather than "clean": @@ -16,14 +17,18 @@ (every call site does `dist.all_reduce(...)`, an attribute lookup on the `torch.distributed` module performed at call time, not at each file's own import time), so this does take effect regardless of load order. -- The cost is real: this changes the process-wide meaning of - `ReduceOp.AVG` for every call site in the process, XPU-related or not. - Anyone reading verl-core's source sees real `ReduceOp.AVG` semantics and - silently gets SUM+divide instead. That is a legitimate argument for fixing - this in verl-core instead (verl-project/verl#7917's - `is_reduce_avg_supported()` hook covers exactly these 3 call sites) -- this - patch exists to show it is *possible* without core changes, not to claim - it is the better design. +- The cost: this still intercepts a public torch API process-wide -- once + applied, every `dist.all_reduce(op=AVG)` call in the process is routed + through this wrapper, XPU-related or not. It is scoped to xccl groups via + `dist.get_backend(group) != "xccl"`, so a gloo/nccl group in the same + process (e.g. a CPU-only Ray actor's coordination group alongside a GPU + worker's xccl group) is untouched and keeps native `AVG` -- but anyone + reading verl-core's source still can't see, from that source alone, that + an xccl group's `ReduceOp.AVG` silently becomes SUM+divide. That is a + legitimate argument for fixing this in verl-core instead + (verl-project/verl#7917's `is_reduce_avg_supported()` hook covers exactly + these 3 call sites) -- this patch exists to show it is *possible* without + core changes, not to claim it is the better design. async_op=True is intentionally unsupported: correctly dividing the result requires the collective to have already completed, which async_op explicitly @@ -59,7 +64,7 @@ def apply() -> None: original_all_reduce = dist.all_reduce def _patched_all_reduce(tensor, op=dist.ReduceOp.SUM, group=None, async_op=False): - if op != dist.ReduceOp.AVG or async_op: + if op != dist.ReduceOp.AVG or async_op or dist.get_backend(group) != "xccl": return original_all_reduce(tensor, op=op, group=group, async_op=async_op) world_size = dist.get_world_size(group=group) @@ -69,4 +74,4 @@ def _patched_all_reduce(tensor, op=dist.ReduceOp.SUM, group=None, async_op=False dist.all_reduce = _patched_all_reduce _applied = True - logger.info("[verl_hardware_plugin] Patched torch.distributed.all_reduce(op=AVG) for XPU/xccl") + logger.info("[verl_hardware_plugin] Patched torch.distributed.all_reduce(op=AVG) for xccl process groups") From 4b90cecf19ba5bbb16c7584754290fb80b0785c6 Mon Sep 17 00:00:00 2001 From: kahlun Date: Thu, 24 Sep 2026 05:31:57 +0000 Subject: [PATCH 06/11] docs(xpu): correct the reason for the reduce_avg workaround Per kahlun's review-thread investigation on verl-project/verl#7917 (measured on real 4xB60 hardware, not reasoned from source): xccl does not uniformly lack ReduceOp.AVG. oneCCL implements it on its SYCL-kernel execution path but not on its scheduler path; path selection is internal to oneCCL and isn't capability-aware, so an AVG request can abort mid-collective ("average operation is not supported for the scheduler path") rather than reliably succeed or reliably fail. Fix tracked for oneCCL 2022.2 / torch 2.15, not yet available on the stack this plugin targets. Also note the separate, narrower double-division bug on very small messages (intel/torch-xpu-ops#3020) is fixed on that same stack and is not what either the FSDP2 engine override or this patch works around -- easy to conflate the two since both are filed against xccl's AVG support. The blanket "xccl does not support ReduceOp.AVG" phrasing this replaces was accurate enough to justify the workaround but not accurate about why, and was written before this investigation existed. Same mechanism, same correctness -- corrected reasoning only. platform_xpu.py has the same blanket phrasing but predates both this branch and #22; left alone here since fixing it isn't part of this PR's diff. 76 passed. --- verl_hardware_plugin/engines/fsdp_xpu.py | 31 +++++++++++++------ .../patches/reduce_avg_allreduce_patch_xpu.py | 20 +++++++++--- 2 files changed, 37 insertions(+), 14 deletions(-) diff --git a/verl_hardware_plugin/engines/fsdp_xpu.py b/verl_hardware_plugin/engines/fsdp_xpu.py index 95c0f1e7..77f58251 100644 --- a/verl_hardware_plugin/engines/fsdp_xpu.py +++ b/verl_hardware_plugin/engines/fsdp_xpu.py @@ -7,19 +7,27 @@ (e.g., force sum reduction for xccl backend). Why is this engine needed? - The Intel xccl collective communication backend does NOT support - ReduceOp.AVG for allreduce operations. FSDP's gradient synchronization + Intel's oneCCL (xccl) collective library implements ReduceOp.AVG on its + SYCL-kernel execution path, but not on its scheduler path -- and which + path a given collective takes is selected internally by oneCCL, not + capability-aware, so an AVG request can abort mid-collective + (`average operation is not supported for the scheduler path`) rather + than reliably succeed or reliably fail. FSDP's gradient synchronization normally uses AVG for efficiency. This engine forces sum-based reduction - followed by manual division, which is functionally equivalent but - compatible with xccl. + followed by manual division, which is functionally equivalent and, + unlike AVG, not subject to that path-selection abort. (Fix tracked for + oneCCL 2022.2 / torch 2.15; not yet available on the stack this plugin + targets. A separate, narrower double-division bug on very small + messages, intel/torch-xpu-ops#3020, is fixed on that same stack and is + not what this workaround is for.) Why not patch verl-core's apply_fsdp2() instead? verl/workers/engine/fsdp/transformer_impl.py does `from verl.utils.fsdp_utils import apply_fsdp2` at its own import time, so a plugin that reassigns `verl.utils.fsdp_utils.apply_fsdp2` after that module has already imported the name would have no effect on this call site -- - the same import-order fragility as verl core's profiler-marker dispatch - (see docs/design/ for that one). This engine sidesteps the problem + the same import-order fragility that motivates verl core's own + profiler-marker dispatch design. This engine sidesteps the problem entirely: it does not patch anything. It calls set_force_sum_reduction_for_comms() on the already-constructed model *after* super().initialize() (which internally calls the real @@ -61,7 +69,8 @@ def _force_sum_reduction_on_all_fsdp_modules(root) -> None: root model (verl/utils/fsdp_utils.py::apply_fsdp2), so each of those per-layer wrappers has its own communication group that also defaults to ReduceOp.AVG. Only fixing the root leaves per-layer gradient sync using - AVG, which xccl silently cannot execute correctly. + AVG, which is subject to oneCCL's unreliable AVG path selection (see + module docstring) instead of the guaranteed-correct SUM+divide path. """ if hasattr(root, "set_force_sum_reduction_for_comms"): root.set_force_sum_reduction_for_comms(True) @@ -97,12 +106,14 @@ def initialize(self): """Initialize the FSDP model, then apply XPU-specific workarounds. The key workaround: force sum-based gradient reduction, root module - and every per-layer FSDP2 wrapper. This is needed because xccl does - not support ReduceOp.AVG. The FSDP wrapper will use SUM + manual + and every per-layer FSDP2 wrapper. This is needed because oneCCL's + AVG path selection isn't capability-aware and can abort mid-collective + (see module docstring). The FSDP wrapper will use SUM + manual division instead. """ super().initialize() - # xccl does not support ReduceOp.AVG; force sum-based reduction + # oneCCL's AVG path selection isn't capability-aware and can abort; + # force sum-based reduction instead (see module docstring) _force_sum_reduction_on_all_fsdp_modules(self.module) diff --git a/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py b/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py index 3e50c7c1..683dc179 100644 --- a/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py +++ b/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py @@ -2,8 +2,19 @@ # Licensed under the Apache License, Version 2.0. """Monkeypatch torch.distributed.all_reduce(op=AVG) -> SUM + manual divide, -scoped to xccl process groups, for Intel XPU (xccl does not implement -ReduceOp.AVG). +scoped to xccl process groups, for Intel XPU. + +Why this is needed: Intel's oneCCL (xccl) implements ReduceOp.AVG on its +SYCL-kernel execution path, but not on its scheduler path. Which path a +given collective takes is selected internally by oneCCL and isn't +capability-aware, so a caller can't tell in advance whether an AVG request +will succeed or abort mid-collective (`average operation is not supported +for the scheduler path`). Forcing sum-based reduction + a manual divide is +the only way to guarantee correctness until that dispatch becomes +capability-aware -- fix tracked for oneCCL 2022.2 / torch 2.15, not yet +available on the stack this plugin targets. (A separate, narrower +double-division bug on very small messages, intel/torch-xpu-ops#3020, is +fixed on that same stack and is not what this patch works around.) This is the one patch in this package worth calling "doable, but with a real design smell" rather than "clean": @@ -33,8 +44,9 @@ async_op=True is intentionally unsupported: correctly dividing the result requires the collective to have already completed, which async_op explicitly defers. Call sites using async_op=True with op=AVG fall through to the -original (broken-on-xccl) behavior rather than silently producing a wrong -answer with no error. +original behavior -- subject to the same oneCCL path-selection risk this +patch exists to avoid -- rather than silently producing a wrong answer with +no error. """ import logging From 07aa5dfe5ed6f295c9c6d6c5b572d5e15865fe86 Mon Sep 17 00:00:00 2001 From: kahln Date: Thu, 24 Sep 2026 20:34:17 -0700 Subject: [PATCH 07/11] fix(tests): simplify mock patching in test_platform_xpu_init_applies_patches --- tests/test_plugin_registration.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_plugin_registration.py b/tests/test_plugin_registration.py index f36f875d..be51e626 100644 --- a/tests/test_plugin_registration.py +++ b/tests/test_plugin_registration.py @@ -706,9 +706,7 @@ class TestReduceAvgPatchWiring: def test_platform_xpu_init_applies_patches(self): from verl_hardware_plugin.platforms.platform_xpu import PlatformXPU - with mock.patch( - "verl_hardware_plugin.patches.reduce_avg_allreduce_patch_xpu.apply" - ) as fake_apply: + with mock.patch("verl_hardware_plugin.patches.reduce_avg_allreduce_patch_xpu.apply") as fake_apply: PlatformXPU() fake_apply.assert_called_once() From b9c4b7d7d42b65e36a5296e4ffb53679292bbb97 Mon Sep 17 00:00:00 2001 From: kahlun Date: Fri, 25 Sep 2026 09:43:37 +0000 Subject: [PATCH 08/11] feat(xpu): move the monkey patch of the avg of xpu to folder --- .../reduce_avg_allreduce_patch.py} | 0 verl_hardware_plugin/platforms/platform_xpu.py | 4 ++-- 2 files changed, 2 insertions(+), 2 deletions(-) rename verl_hardware_plugin/patches/{reduce_avg_allreduce_patch_xpu.py => xpu/reduce_avg_allreduce_patch.py} (100%) diff --git a/verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py b/verl_hardware_plugin/patches/xpu/reduce_avg_allreduce_patch.py similarity index 100% rename from verl_hardware_plugin/patches/reduce_avg_allreduce_patch_xpu.py rename to verl_hardware_plugin/patches/xpu/reduce_avg_allreduce_patch.py diff --git a/verl_hardware_plugin/platforms/platform_xpu.py b/verl_hardware_plugin/platforms/platform_xpu.py index 3c095865..08d2516e 100644 --- a/verl_hardware_plugin/platforms/platform_xpu.py +++ b/verl_hardware_plugin/platforms/platform_xpu.py @@ -89,9 +89,9 @@ def __init__(self) -> None: # host with VERL_PLATFORM=nvidia explicitly set would still get its # process-wide torch.distributed.all_reduce(op=AVG) semantics # changed for no reason. - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch - reduce_avg_allreduce_patch_xpu.apply() + reduce_avg_allreduce_patch.apply() # ------------------------------------------------------------------ # Core device management From cac78f9582ac88e30b441b5b5cf65dc09d2bb3a7 Mon Sep 17 00:00:00 2001 From: kahlun Date: Mon, 28 Sep 2026 01:18:21 +0000 Subject: [PATCH 09/11] fix(patches): resolve pytest update import paths for reduce_avg_allreduce_patch in xpu module --- tests/test_plugin_registration.py | 16 ++++++++-------- verl_hardware_plugin/patches/__init__.py | 4 ++-- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/tests/test_plugin_registration.py b/tests/test_plugin_registration.py index be51e626..9682a7c5 100644 --- a/tests/test_plugin_registration.py +++ b/tests/test_plugin_registration.py @@ -706,7 +706,7 @@ class TestReduceAvgPatchWiring: def test_platform_xpu_init_applies_patches(self): from verl_hardware_plugin.platforms.platform_xpu import PlatformXPU - with mock.patch("verl_hardware_plugin.patches.reduce_avg_allreduce_patch_xpu.apply") as fake_apply: + with mock.patch("verl_hardware_plugin.patches.xpu.reduce_avg_allreduce_patch.apply") as fake_apply: PlatformXPU() fake_apply.assert_called_once() @@ -733,7 +733,7 @@ class TestReduceAvgAllReducePatch: def _reset_patch_state(self): import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod original_all_reduce = dist.all_reduce patch_mod._applied = False @@ -744,7 +744,7 @@ def _reset_patch_state(self): def test_noop_when_xpu_unavailable(self): import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod before = dist.all_reduce with mock.patch.object(patch_mod, "_xpu_available", return_value=False): @@ -756,7 +756,7 @@ def test_avg_becomes_sum_plus_divide(self): import torch import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod fake_original = mock.MagicMock() tensor = torch.tensor([4.0]) @@ -778,7 +778,7 @@ def test_non_xccl_backend_passes_through_even_with_avg(self): import torch import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod fake_original = mock.MagicMock() tensor = torch.tensor([4.0]) @@ -794,7 +794,7 @@ def test_non_avg_op_passes_through_unchanged(self): import torch import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod fake_original = mock.MagicMock() tensor = torch.tensor([4.0]) @@ -809,7 +809,7 @@ def test_async_avg_falls_through_unpatched(self): import torch import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod fake_original = mock.MagicMock() tensor = torch.tensor([4.0]) @@ -823,7 +823,7 @@ def test_async_avg_falls_through_unpatched(self): def test_idempotent(self): import torch.distributed as dist - from verl_hardware_plugin.patches import reduce_avg_allreduce_patch_xpu as patch_mod + from verl_hardware_plugin.patches.xpu import reduce_avg_allreduce_patch as patch_mod with mock.patch.object(patch_mod, "_xpu_available", return_value=True): patch_mod.apply() diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py index 6f9d1d8c..b9a0ea34 100644 --- a/verl_hardware_plugin/patches/__init__.py +++ b/verl_hardware_plugin/patches/__init__.py @@ -12,9 +12,9 @@ the corresponding hardware/SDK merely happens to be importable, regardless of which platform verl actually selects for the run. -Currently contains one module: +Currently contains one platform subpackage: -- `reduce_avg_allreduce_patch_xpu`: Intel XPU's `is_reduce_avg_supported` +- `xpu.reduce_avg_allreduce_patch`: Intel XPU's `is_reduce_avg_supported` replacement. See its own docstring for what it does and why it's a monkeypatch rather than a `PlatformBase` hook (verl-hardware-plugin#26). """ From dd71d54c418b29e8104ab6a79e73fe89ad0b00d7 Mon Sep 17 00:00:00 2001 From: kahlun Date: Wed, 30 Sep 2026 03:08:22 +0000 Subject: [PATCH 10/11] docs(patches): clarify the application context of monkeypatches in patches/__init__.py --- verl_hardware_plugin/patches/__init__.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py index b9a0ea34..53dfed40 100644 --- a/verl_hardware_plugin/patches/__init__.py +++ b/verl_hardware_plugin/patches/__init__.py @@ -5,16 +5,19 @@ the platform class that needs them -- not from this package's `__init__.py` and not from this plugin's shared top-level `verl_hardware_plugin/__init__.py`. -`PlatformXXX()` is only constructed once `verl.plugin.platform. +Each `PlatformXXX()` is only constructed once `verl.plugin.platform. platform_manager._create_platform()` has actually selected that platform for -the process (see `PlatformXPU.__init__` in `platforms/platform_xpu.py`). -Applying a patch from a shared location instead would fire on any host where -the corresponding hardware/SDK merely happens to be importable, regardless of -which platform verl actually selects for the run. +the process, so a platform's own `__init__` is the right place to apply its +patches -- not this package, and not the plugin's shared top-level +`__init__.py`. Applying a patch from a shared location instead would fire on +any host where the corresponding hardware/SDK merely happens to be +importable, regardless of which platform verl actually selects for the run. Currently contains one platform subpackage: - `xpu.reduce_avg_allreduce_patch`: Intel XPU's `is_reduce_avg_supported` - replacement. See its own docstring for what it does and why it's a - monkeypatch rather than a `PlatformBase` hook (verl-hardware-plugin#26). + replacement, applied from `PlatformXPU.__init__` + (`platforms/platform_xpu.py`). See its own docstring for what it does and + why it's a monkeypatch rather than a `PlatformBase` hook + (verl-hardware-plugin#26). """ From ff5416ec66cff5342c488b40be60e460c8915fcd Mon Sep 17 00:00:00 2001 From: kahlun Date: Wed, 30 Sep 2026 03:14:20 +0000 Subject: [PATCH 11/11] docs(patches): streamline documentation in __init__.py by removing outdated platform subpackage references --- verl_hardware_plugin/patches/__init__.py | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/verl_hardware_plugin/patches/__init__.py b/verl_hardware_plugin/patches/__init__.py index 53dfed40..397f1c84 100644 --- a/verl_hardware_plugin/patches/__init__.py +++ b/verl_hardware_plugin/patches/__init__.py @@ -8,16 +8,7 @@ Each `PlatformXXX()` is only constructed once `verl.plugin.platform. platform_manager._create_platform()` has actually selected that platform for the process, so a platform's own `__init__` is the right place to apply its -patches -- not this package, and not the plugin's shared top-level -`__init__.py`. Applying a patch from a shared location instead would fire on +patches. Applying a patch from a shared location instead would fire on any host where the corresponding hardware/SDK merely happens to be importable, regardless of which platform verl actually selects for the run. - -Currently contains one platform subpackage: - -- `xpu.reduce_avg_allreduce_patch`: Intel XPU's `is_reduce_avg_supported` - replacement, applied from `PlatformXPU.__init__` - (`platforms/platform_xpu.py`). See its own docstring for what it does and - why it's a monkeypatch rather than a `PlatformBase` hook - (verl-hardware-plugin#26). """