experiment: pure-monkeypatch XPU integration, zero PlatformBase hooks

docs/design/xpu-monkeypatch-experiment.md

@@ -0,0 +1,122 @@
+# Experiment: how much of #7917 can live in the plugin as 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 is the answer to "then how much of #7917 actually can't be done
+from here" — implemented, not just argued. It does not touch
+`feature/xpu-vtune-avg-attention` (the branch behind open PR #22, which
+still uses the hook-based approach and depends on #7917 merging). This is a
+separate, from-`main` branch so the two can be compared directly for size
+and how much they depend on core changes ever landing.
+
+## Verdict per capability
+
+| Capability | In this branch? | Mechanism | Notes |
+|---|---|---|---|
+| `attention_utils_module` | Yes | 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) | Yes | 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. |
+| `is_reduce_avg_supported` (3 raw `dist.all_reduce(op=AVG)` sites) | Yes, flagged | 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`. |
+| `dist_profiler_cls` (VTune tool selection) | Yes | Patch `DistProfiler.__init__` (mutate the class method, not the name) | Clean — see `patches/dist_profiler_patch_xpu.py` docstring for why patching the method and not reassigning the class name matters. |
+| `profiler_markers` (ambient `marked_timer`/`mark_start_range` calls throughout the trainer) | **No** | — | Genuine blocker. `verl/utils/profiler/__init__.py` picks one of three whole modules with an `if/elif/else` **once**, at whatever moment it's first imported by *anything* in the process — there's no per-call dispatch object to patch. Reassigning `verl.utils.profiler.marked_timer` after the fact only helps callers that read that name *after* the patch runs; anything that already did `from verl.utils.profiler import marked_timer` (or `from verl.utils.debug import *`, which re-exports it) earlier keeps its own bound reference. Whether this patch would work in practice depends on winning a race against verl-core's own import order — not a foundation to build on. |
+| `dist_profiler_cls` — rollout-server profiler allowlist (vLLM/SGLang/TRT-LLM async servers) | **No** | — | Genuine blocker, same conclusion as `verl-hardware-plugin#26`: `profiler_config = None` and the call that consumes it are in the same method body, with no hook or registry in between. Nothing to intercept short of forking ~40 lines of server constructor per engine. |
+
+Net: 3 of 4 hooks from #7917 are unnecessary — this plugin already does what
+they'd do, with zero core changes. The 4th (`profiler_markers`) turns out to
+have the *same* import-order problem as the rollout-server case, which
+earlier framing (including an earlier version of this repo's own proposal
+doc) had marked as "clean." It isn't. Two things, not one, look like
+legitimate small-core-PR asks now — not the 4 hooks #7917 currently has.
+
+## Why the FSDP2 fix isn't `patches/reduce_avg_fsdp2_patch_xpu.py`
+
+The first draft of this plan (matching the 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` — the exact same class of import-order fragility as
+`profiler_markers` above.
+
+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 just extended 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`), which is
+what `is_reduce_avg_supported`'s FSDP2 wiring in #7917 would have bought —
+achieved here with no patch and no hook.
+
+## Comparison
+
+| | Files touched outside this plugin | Depends on #7917 merging |
+|---|---|---|
+| PR #22 (`feature/xpu-vtune-avg-attention`) | 0 (but 4 `PlatformXPU` hook methods are dead code without #7917) | Yes, for all 4 hooks |
+| This branch | 0 | No, for 3 of 4. `profiler_markers`'s ambient markers and the rollout-server allowlist remain open — either fork ~40 lines of server code into the plugin, or ask for the two small core changes verl-hardware-plugin#26 already discusses. |
+
+## Failure policy for the patches
+
+Every module in `patches/` shares one XPU guard (`patches/_xpu_guard.py`) and
+is a hard no-op on a CPU/CUDA/NPU process — the entry point is discovered on
+every host, not just Intel ones, and all three patches mutate process-global
+state (`torch.distributed.all_reduce`, `verl.utils.attention_utils.
+_get_attention_functions`, `DistProfiler.__init__`). Patching any of those on
+another vendor's host changes *that* vendor's behavior.
+
+`apply_all()` splits failures by consequence rather than swallowing them:
+
+| Patch | On failure, on an XPU host |
+|---|---|
+| `reduce_avg_allreduce_patch_xpu` | **raise** — oneCCL's `ReduceOp.AVG` is not reliably available; continuing risks a mid-collective abort or wrong gradients |
+| `attention_patch_xpu` | **raise** — rmpad attention needs the XPU function set |
+| `dist_profiler_patch_xpu` | `logger.exception`, continue — losing `tool: vtune` costs observability, not correctness |
+
+On a non-XPU host nothing was going to install anyway, so a failure is
+debug-logged and never raised.
+
+## Hardware validation
+
+Run on 2×B60 (Intel Arc Pro B60 Graphics) via devctl, 2026-09-24, inside the
+image built by PR #2's `docker/intel_gpu/Dockerfile.intel_gpu` with this
+branch as the build context (verl-core `cbf4f31b`, torch `2.13.0+xpu`, vLLM
+`0.29.0`):
+
+- 85 unit tests pass in-image.
+- All three patches install on an XPU host; none install in the same image on
+  a CPU-only container (`dist.all_reduce` and `DistProfiler.__init__` verified
+  untouched there).
+- `attention_utils._get_attention_functions()` returns the
+  `verl.utils.npu_flash_attn_utils` set; `profiler.tool: vtune` resolves to
+  `VtuneProfiler`.
+- 2-rank xccl: patched `all_reduce(AVG)` returns the true mean (1.5 for ranks
+  contributing 1 and 2), `SUM` is unaffected.
+- PR #8's premise holds on this image: verl-core `cbf4f31b` *does* define
+  `attention_utils_module`/`profiler_markers`/`dist_profiler_cls` on
+  `PlatformBase`, and this plugin overrides none of them — so the hooks are
+  inert and the patches are what produced the results above.
+
+One caveat worth recording, since it is easy to misread: a plugin-free
+baseline on the same pod showed unpatched xccl computing `ReduceOp.AVG`
+*correctly* (both sync and async) for a small 2-rank tensor. That does not
+contradict the patch — oneCCL implements AVG on its SYCL-kernel path and not
+its scheduler path, and path selection is internal, so a small collective can
+land on the working path. It does mean a 2-rank smoke test cannot demonstrate
+the failure the patch prevents.
+
+## Not yet done
+
+- No end-to-end GRPO training run on this branch; the validation above
+  exercises the patches directly, not a full trainer loop.
+- No unit tests yet for the two capabilities this branch deliberately does
+  *not* patch (they are argued from source, not exercised).

docs/development.md

@@ -577,13 +577,19 @@ class FSDPMyVendorEngineWithLMHead(FSDPEngineWithLMHead):
         - Force sum reduction if the comm backend doesn't support AVG
         - Enable/disable specific FSDP features
         - Apply vendor-specific optimizations
+
+        The wrapped model is `self.module`, assigned by
+        `FSDPEngine._build_model_optimizer()` (called from `initialize()`) in
+        verl/workers/engine/fsdp/transformer_impl.py. There is no `self.model`
+        on these engines -- only `self.model_config` -- so a `hasattr(self.model,
+        ...)` guard here silently never fires.
         """
         super().initialize()
 
         # Example: Force sum-based reduction for custom communication backend
         # (some backends like xccl don't support ReduceOp.AVG)
-        if hasattr(self.model, "set_force_sum_reduction_for_comms"):
-            self.model.set_force_sum_reduction_for_comms(True)
+        if hasattr(self.module, "set_force_sum_reduction_for_comms"):
+            self.module.set_force_sum_reduction_for_comms(True)
             logger.info("Enabled force_sum_reduction for MyVendor comm backend")
 
 
@@ -607,8 +613,8 @@ class FSDPMyVendorEngineWithValueHead(FSDPEngineWithValueHead):
 
     def initialize(self):
         super().initialize()
-        if hasattr(self.model, "set_force_sum_reduction_for_comms"):
-            self.model.set_force_sum_reduction_for_comms(True)
+        if hasattr(self.module, "set_force_sum_reduction_for_comms"):
+            self.module.set_force_sum_reduction_for_comms(True)
 ```
 
 Then register in `verl_hardware_plugin/engines/__init__.py`:
@@ -730,8 +736,8 @@ And the corresponding engine:
 class FSDPXPUEngineWithLMHead(FSDPEngineWithLMHead):
     def initialize(self):
         super().initialize()
-        if hasattr(self.model, "set_force_sum_reduction_for_comms"):
-            self.model.set_force_sum_reduction_for_comms(True)
+        if hasattr(self.module, "set_force_sum_reduction_for_comms"):
+            self.module.set_force_sum_reduction_for_comms(True)
 ```
 
 ---
@@ -849,7 +855,7 @@ Environment variable overrides:
 
 **Solution**: In your engine's `initialize()`, enable force sum reduction:
 ```python
-self.model.set_force_sum_reduction_for_comms(True)
+self.module.set_force_sum_reduction_for_comms(True)
 ```
 
 ### Problem: Import errors when plugin is installed on machines without my hardware

tests/test_fsdp_xpu_engine.py

@@ -0,0 +1,77 @@
+# Copyright (c) 2026 BAAI. All rights reserved.
+# Licensed under the Apache License, Version 2.0.
+
+"""Regression tests for the Intel XPU FSDP engine's xccl reduce workaround.
+
+Two things are pinned here:
+
+1. The attribute name. verl's FSDPEngine assigns the wrapped model to
+   ``self.module`` (transformer_impl.py::_build_model_optimizer, called from
+   ``initialize()``); there is no ``self.model`` on these engines, only
+   ``self.model_config``.
+   The engine previously guarded on ``hasattr(self.model, ...)``, so the
+   workaround silently never ran and xccl kept getting ReduceOp.AVG.
+2. The recursion. fully_shard() wraps each transformer layer as well as the
+   root, and each wrapper has its own reduction op, so fixing only the root
+   leaves per-layer gradient sync on AVG.
+"""
+
+import pytest
+
+torch = pytest.importorskip("torch")
+nn = torch.nn
+
+
+@pytest.fixture(scope="module")
+def force_sum_helper():
+    fsdp_xpu = pytest.importorskip("verl_hardware_plugin.engines.fsdp_xpu")
+    return fsdp_xpu._force_sum_reduction_on_all_fsdp_modules
+
+
+class _FakeFSDPModule(nn.Module):
+    """Stands in for a fully_shard()-wrapped module: has the setter, records calls."""
+
+    def __init__(self):
+        super().__init__()
+        self.forced = None
+
+    def set_force_sum_reduction_for_comms(self, value):
+        self.forced = value
+
+
+class _FakeFSDPRoot(_FakeFSDPModule):
+    def __init__(self, n_layers=3):
+        super().__init__()
+        self.layers = nn.ModuleList([_FakeFSDPModule() for _ in range(n_layers)])
+        # A plain submodule with no setter must simply be skipped, not crash.
+        self.embed = nn.Linear(4, 4)
+
+
+def test_force_sum_reduction_covers_root_and_every_wrapped_layer(force_sum_helper):
+    root = _FakeFSDPRoot(n_layers=3)
+
+    force_sum_helper(root)
+
+    assert root.forced is True
+    assert [layer.forced for layer in root.layers] == [True, True, True]
+
+
+def test_force_sum_reduction_tolerates_a_model_without_the_setter(force_sum_helper):
+    """FSDP1 / unwrapped models have no setter at all; this must be a no-op."""
+    force_sum_helper(nn.Linear(4, 4))
+
+
+def test_verl_fsdp_engine_exposes_module_not_model():
+    """Guards the attribute contract the workaround depends on.
+
+    If a future verl release renames ``module``, this fails here instead of
+    silently disabling the workaround at runtime.
+    """
+    import inspect
+
+    transformer_impl = pytest.importorskip("verl.workers.engine.fsdp.transformer_impl")
+    # Scan the whole class, not initialize(): the assignment lives in
+    # _build_model_optimizer(), which initialize() calls.
+    class_src = inspect.getsource(transformer_impl.FSDPEngine)
+    assert "self.module = " in class_src, "verl's FSDPEngine no longer assigns self.module"
+    assert "self.model = " not in class_src, "verl's FSDPEngine now also assigns self.model; revisit the engine"

tests/test_itt_profile_xpu.py

@@ -0,0 +1,111 @@
+# Copyright (c) 2026 BAAI. All rights reserved.
+# Licensed under the Apache License, Version 2.0.
+
+"""Regression tests for the Intel VTune (ITT) profiling backend.
+
+ITT ranges are a push/pop stack per thread. A range that is pushed but never
+popped does not just lose one measurement -- every later range on that thread
+nests inside the leaked one, so the trace is unreadable exactly in the runs
+where something went wrong and you most want to read it. These tests pin the
+push/pop balance when the timed body raises.
+"""
+
+from unittest import mock
+
+import pytest
+
+from verl_hardware_plugin.profilers import itt_profile_xpu
+
+
+class _ITTSpy:
+    """Stand-in for torch.profiler.itt that tracks push/pop depth."""
+
+    def __init__(self):
+        self.depth = 0
+        self.messages = []
+
+    def range_push(self, message):
+        self.messages.append(message)
+        self.depth += 1
+        return self.depth
+
+    def range_pop(self):
+        self.depth -= 1
+        return self.depth
+
+
+@pytest.fixture
+def itt():
+    spy = _ITTSpy()
+    with mock.patch.object(itt_profile_xpu, "_itt", spy):
+        yield spy
+
+
+def test_marked_timer_pops_range_on_success(itt):
+    timing = {}
+    with itt_profile_xpu.marked_timer("step", timing):
+        pass
+    assert itt.depth == 0
+    assert itt.messages == ["step"]
+    assert "step" in timing
+
+
+def test_marked_timer_pops_range_when_body_raises(itt):
+    timing = {}
+    with pytest.raises(ValueError, match="boom"):
+        with itt_profile_xpu.marked_timer("step", timing):
+            raise ValueError("boom")
+    assert itt.depth == 0, "a raising body must not leave the ITT range open"
+
+
+def test_marked_timer_propagates_the_original_exception(itt):
+    """The timing exception must survive the finally, not be replaced by it."""
+    with pytest.raises(KeyError):
+        with itt_profile_xpu.marked_timer("step", {}):
+            raise KeyError("original")
+
+
+def test_annotate_pops_range_when_decorated_function_raises(itt):
+    profiler = itt_profile_xpu.VtuneProfiler(rank=0, config=None, tool_config=None)
+
+    @profiler.annotate(message="rollout")
+    def boom():
+        raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        boom()
+    assert itt.depth == 0, "a raising decorated function must not leave the ITT range open"
+    assert itt.messages == ["rollout"]
+
+
+def test_annotate_returns_value_and_pops_range_on_success(itt):
+    profiler = itt_profile_xpu.VtuneProfiler(rank=0, config=None, tool_config=None)
+
+    @profiler.annotate()
+    def add(a, b):
+        return a + b
+
+    assert add(1, 2) == 3
+    assert itt.depth == 0
+    assert itt.messages == ["add"]
+
+
+def test_annotate_stops_discrete_collection_when_function_raises(itt):
+    """With discrete=True the platform profiler is started per call; it must be
+    stopped on the error path too, or the collector keeps running forever."""
+    profiler = itt_profile_xpu.VtuneProfiler(rank=0, config=None, tool_config=None)
+    profiler.discrete = True
+
+    platform = mock.MagicMock()
+    with mock.patch.object(itt_profile_xpu, "get_platform", return_value=platform):
+
+        @profiler.annotate()
+        def boom():
+            raise RuntimeError("boom")
+
+        with pytest.raises(RuntimeError):
+            boom()
+
+    platform.profiler_start.assert_called_once()
+    platform.profiler_stop.assert_called_once()
+    assert itt.depth == 0