[c10d][xccl2] Port the torchcomms XCCL integration suite onto xccl2

[frost-intel]

Stack (bottom → top)

	PR	Branch
	#5	xccl2/01-foundation
	#6	xccl2/02-api
	#7	xccl2/03-work
	#8	xccl2/04-engine-decl
	#9	xccl2/05-engine-impl
	#10	xccl2/06-backend-surface
	#11	xccl2/07-build-wiring
	#12	xccl2/08-tests
→	#3	xccl2/09-integration-tests

Each PR is exactly one commit and targets the one below it. Review bottom-up.

---

torchcomms ships 36 Python integration tests for its XCCL backend
(comms/torchcomms/scripts/run_tests_integration_xccl_py.sh). They are the
acceptance suite that backend was developed against, and none of them ran
against xccl2. This adds ports of all 36 under
test/distributed/xccl2_integration/, so the in-tree backend carries the same
coverage.

These are not copies. xccl2 is a c10d Backend, reachable only through
dist.init_process_group(backend="xccl2"); there is no TorchComm object and no
new_comm("xccl2", ...). So each port is an API translation, and the originals
fall into three groups:

A. Already c10d-native. These set dist.config.use_torchcomms = True and then
use plain dist.* calls. Porting is dropping that flag and passing
backend="xccl2". Notably the three BackendWrapper* tests land here --
BackendWrapper is the shim that makes a TorchComm look like a c10d
Backend, so its tests were written against pure c10d already.

B. Written against the TorchComm API (self.torchcomm.all_reduce(t, op,
async_op)). Rewritten against the dist.* equivalent, preserving the
operand construction, the sweep over counts/dtypes/ops, and the
verification.

C. Reaching into torchcomms internals (options, finalize, mempool, custom
ops). Reduced to what c10d exposes, or marked as a parity gap.

common.py holds the shared harness: process-group setup/teardown per test
class, rank/size discovery across OMPI/SLURM/PMI/PALS launchers, dtype and op
naming, int8 overflow filtering, tensor verification, and
CollectiveVariantsMixin, which factors out the five call shapes every
torchcomms collective test repeats (sync, sync-no-work, async,
async-early-reset, input-deleted). The originals' CUDA-graph variants are
ncclx-only and are not ported.

Tests covering functionality xccl2 does not yet implement are decorated with
@parity_gap so they are reported as skips naming the missing piece rather than
silently omitted:

• premul_sum oneCCL lowers PREMUL_SUM to SUM (oneCCL#195, Add more optimizers pytorch/pytorch#196)

MemPoolTest runs rather than skips, now that ProcessGroupXCCL implements
getMemAllocator(). It also needed a fix the original could not have caught: it
called backend.get_mem_allocator(), which does not exist on any c10d backend.
The binding is a property, Backend.mem_allocator, so the port uses that, the
same way the nccl2 tests do.

AllGatherVTest and SplitTest run rather than skip. SplitTest drives
dist.split_group() rather than dist.new_group(): new_group() never reaches
Backend::split, and the torchcomms original has each rank name only its own
half, which for new_group() is a different ranks list per rank and diverges the
store rendezvous. split_group() is collective over the parent and takes the
same split_ranks everywhere, so it is the honest c10d equivalent of
TorchComm.split(). It also requires the default group to be bound to a device,
which the shared harness does not do, so SplitTest binds one in setUpClass.

run_tests_integration_xccl2_py.sh mirrors the torchcomms runner, driving each
file under torchrun. TEST_BACKEND, TEST_DEVICE and TEST_FULL_SWEEP override the
backend string, device type and sweep size.

Validated on 12 XPUs (4 ranks): all 36 files pass. Notes from that run:

• The async send/recv variants follow the original's rank-parity ordering.
  Ungrouped bidirectional p2p deadlocks on oneCCL -- stock xccl hangs on the
  same pattern -- which is why batch_isend_irecv coalescing exists.

• broadcast/reduce/barrier are marked @dynamo_gap rather than @parity_gap:
  the dist.* wrappers construct pybind11 options objects that dynamo cannot
  trace (gb0156), so they fail under fullgraph=True on every c10d backend.
  all_reduce, all_gather(_into_tensor), reduce_scatter_tensor and
  all_to_all_single do trace, and are asserted to compile into one graph.

• abort() and repeated destroy are not tested: abortXcclComm() calls ::abort()
  unconditionally (abort_process_on_timeout_or_error_ is hardcoded true), and
  c10d rejects a repeated destroy at the Python layer. The shared-comm
  double-shutdown the original covers is reached via the mixed cpu:gloo,
  xpu:xccl2 group instead.

MultiCommTest found a real use-after-free, which is fixed earlier in this stack
rather than here: a work handle still held by Python when
destroy_process_group() runs would return its XPU events to the backend's
event pool after the backend had been destroyed. The work no longer points at
its backend; see "Port the XCCL work object, work queue and bootstrap".

All 36 files pass on 4 ranks of PVC, with 6 skips: the premul_sum parity gap
above, wait_blocking, and the four dynamo ones.