Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -498,6 +498,7 @@ jobs:
pytest -rsf fbgemm_gpu/test/jagged/2d_to_dense_test.py
pytest -rsf fbgemm_gpu/test/jagged/1d_to_dense_test.py
pytest -rsf fbgemm_gpu/test/jagged/elementwise_binary_test.py
pytest -rsf fbgemm_gpu/test/merge_pooled_embeddings_test.py
pytest -rsf fbgemm_gpu/test/sparse/block_bucketize_test.py
pytest -rsf fbgemm_gpu/test/sparse/cumsum_test.py
pytest -rsf fbgemm_gpu/test/sparse/misc_ops_test.py
Expand Down
5 changes: 5 additions & 0 deletions packages/fbgemm-xpu/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ operators alongside the operators above. You can find their exact signature in
- `populate_bucketized_permute`
- `reorder_batched_ad_indices`
- `reorder_batched_ad_lengths`
- `all_to_one_device`

## Supported hardware

Expand Down Expand Up @@ -154,6 +155,10 @@ FBGEMM tests cover the existing non-lookup operators.
`bounds_check_indices` implements version 1 only. Version 2 and
`prefetch_pipeline=True` are not implemented on XPU.

`all_to_one_device` copies between XPU devices of one process. The pooled
merge operators (`merge_pooled_embeddings`, `sum_reduce_to_one`) are not
implemented on XPU.

[FBGEMM]: https://github.com/pytorch/FBGEMM
[uv]: https://github.com/astral-sh/uv
[PVC]: https://www.intel.com/content/www/us/en/ark/products/series/232874/intel-data-center-gpu-max-series.html
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
From a3f79329fc193afe34ce69feb1b6690b07aaf618 Mon Sep 17 00:00:00 2001
From 0c0d29b0926c2c93b8564441b1e1040684d71f3a Mon Sep 17 00:00:00 2001
From: Alberto Gallegos <alberto.gallegos.muro@intel.com>
Date: Thu, 9 Jul 2026 18:28:28 +0000
Subject: [PATCH] Add XPU support to FBGEMM sparse ops and jagged tensor tests
Expand Down Expand Up @@ -55,8 +55,15 @@ add_jagged_output rather than skipping the tests. All three then run and keep
covering the implemented operator, including the vectorized Half fast path and the
torch.compile dynamic-shape path.

Enable test_all_to_one_device in merge_pooled_embeddings_test.py on the
current accelerator through the shared accelerator helpers. The class-level
gpu_unavailable skip moves to the individual tests, so the remaining tests,
which cover merge_pooled_embeddings and sum_reduce_to_one (no XPU
implementation), keep running on CUDA and skip automatically on XPU.

Signed-off-by: Alberto Gallegos <alberto.gallegos.muro@intel.com>
Signed-off-by: Felipe Leza Alvarez <felipe.leza.alvarez@intel.com>
Signed-off-by: mkrze <mateusz.krzemieniewski@intel.com>

Co-authored-by: Felipe Leza Alvarez <felipe.leza.alvarez@intel.com>
---
Expand All @@ -67,6 +74,7 @@ Co-authored-by: Felipe Leza Alvarez <felipe.leza.alvarez@intel.com>
.../test/jagged/elementwise_binary_test.py | 19 +-
.../jagged/expand_into_jagged_permute_test.py | 29 +--
.../jagged/jagged_index_select_2d_test.py | 28 ++-
.../test/merge_pooled_embeddings_test.py | 31 +++-
.../test/sparse/block_bucketize_test.py | 172 ++++++++++--------
fbgemm_gpu/test/sparse/cumsum_test.py | 4 +
fbgemm_gpu/test/sparse/misc_ops_test.py | 24 ++-
Expand All @@ -75,7 +83,7 @@ Co-authored-by: Felipe Leza Alvarez <felipe.leza.alvarez@intel.com>
.../tbe/utils/generate_vbe_metadata_test.py | 23 ++-
.../tbe/utils/split_embeddings_utils_test.py | 33 +++-
fbgemm_gpu/test/test_utils.py | 52 +++++-
15 files changed, 368 insertions(+), 166 deletions(-)
16 files changed, 392 insertions(+), 173 deletions(-)

diff --git a/fbgemm_gpu/test/jagged/1d_to_dense_test.py b/fbgemm_gpu/test/jagged/1d_to_dense_test.py
index 387ee1010..9ceb293fb 100644
Expand Down Expand Up @@ -509,6 +517,129 @@ index 2e28863ad..4cd048eb0 100644
is_float = jagged_tensor_dtype in [torch.float, torch.half, torch.bfloat16]
lengths = torch.randint(
low=0,
diff --git a/fbgemm_gpu/test/merge_pooled_embeddings_test.py b/fbgemm_gpu/test/merge_pooled_embeddings_test.py
index f0db0e65b..4326f4d83 100644
--- a/fbgemm_gpu/test/merge_pooled_embeddings_test.py
+++ b/fbgemm_gpu/test/merge_pooled_embeddings_test.py
@@ -21,14 +21,22 @@ open_source: bool = getattr(fbgemm_gpu, "open_source", False)

if open_source:
# pyre-ignore[21]
- from test_utils import gpu_unavailable
+ from test_utils import accelerator_unavailable, current_accelerator, gpu_unavailable
else:
import fbgemm_gpu.sparse_ops # noqa: F401, E402
- from fbgemm_gpu.test.test_utils import gpu_unavailable
+ from fbgemm_gpu.test.test_utils import (
+ accelerator_unavailable,
+ current_accelerator,
+ gpu_unavailable,
+ )

torch.ops.load_library("//deeplearning/fbgemm/fbgemm_gpu:merge_pooled_embeddings")

+# Registers the XPU (Intel GPU) FBGEMM operator implementations.
+import fbgemm_xpu # noqa: F401, E402
+
typed_gpu_unavailable: tuple[bool, str] = gpu_unavailable
+typed_accelerator_unavailable: tuple[bool, str] = accelerator_unavailable


def make_pitched_tensor(
@@ -53,11 +61,11 @@ def make_pitched_tensor(


# @unittest.skipIf(open_source, "Not supported in open source yet")
-@unittest.skipIf(*typed_gpu_unavailable)
class MergePooledEmbeddingsTest(unittest.TestCase):
# pyre-fixme[56]: Pyre was not able to infer the type of argument
# `hypothesis.strategies.integers($parameter$min_value = 1, $parameter$max_value =
# 10)` to decorator factory `hypothesis.given`.
+ @unittest.skipIf(*typed_gpu_unavailable)
@given(
num_ads=st.integers(min_value=1, max_value=10),
embedding_dimension=st.integers(min_value=1, max_value=32),
@@ -126,9 +134,10 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
torch.testing.assert_close(output_ref, output_cpu)

# pyre-fixme[56]: Pyre was not able to infer the type of argument
+ @unittest.skipIf(*typed_accelerator_unavailable)
@given(
num_inputs=st.integers(min_value=1, max_value=10),
- num_gpus=st.integers(min_value=1, max_value=torch.cuda.device_count()),
+ num_gpus=st.integers(min_value=1, max_value=torch.accelerator.device_count()),
r=st.randoms(use_true_random=False),
use_pitched=st.booleans(),
)
@@ -142,8 +151,9 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
r,
use_pitched: bool,
) -> None:
- dst_device = torch.device(f"cuda:{r.randint(0, num_gpus - 1)}")
- with torch.cuda.device(dst_device):
+ device_type = current_accelerator().type
+ dst_device = torch.device(device_type, r.randint(0, num_gpus - 1))
+ with torch.accelerator.device_index(dst_device.index):
if use_pitched:
inputs = [
make_pitched_tensor(10, 20, torch.float32, "cpu", alignment=256)
@@ -153,13 +163,15 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
inputs = [torch.randn(10, 20) for _ in range(num_inputs)]

cuda_inputs = [
- input.to(f"cuda:{i % num_gpus}") for i, input in enumerate(inputs)
+ input.to(torch.device(device_type, i % num_gpus))
+ for i, input in enumerate(inputs)
]
cuda_outputs = torch.ops.fbgemm.all_to_one_device(cuda_inputs, dst_device)
for i, o in zip(inputs, cuda_outputs):
self.assertEqual(o.device, dst_device)
torch.testing.assert_close(o.cpu(), i)

+ @unittest.skipIf(*typed_gpu_unavailable)
def test_merge_pooled_embeddings_gpu_to_cpu(self) -> None:
dst_device = torch.device("cpu")
inputs = [torch.randn(10, 20) for _ in range(4)]
@@ -174,6 +186,7 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
torch.testing.assert_close(output, ref_output)

# pyre-fixme[56]: Pyre was not able to infer the type of argument
+ @unittest.skipIf(*typed_gpu_unavailable)
@given(
num_inputs=st.integers(min_value=1, max_value=8),
num_gpus=st.integers(min_value=1, max_value=torch.cuda.device_count()),
@@ -213,6 +226,7 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
)
torch.testing.assert_close(output, ref_output)

+ @unittest.skipIf(*typed_gpu_unavailable)
def test_merge_pooled_embeddings_cpu_with_different_target_device(self) -> None:
uncat_size = 2
pooled_embeddings = [torch.ones(uncat_size, 4), torch.ones(uncat_size, 8)]
@@ -228,6 +242,7 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
# pyre-fixme[56]: Pyre was not able to infer the type of argument
# `hypothesis.strategies.integers($parameter$min_value = 1, $parameter$max_value =
# 10)` to decorator factory `hypothesis.given`.
+ @unittest.skipIf(*typed_gpu_unavailable)
@given(
num_inputs=st.integers(min_value=1, max_value=10),
num_gpus=st.integers(min_value=1, max_value=torch.cuda.device_count()),
@@ -256,6 +271,7 @@ class MergePooledEmbeddingsTest(unittest.TestCase):
cuda_output.cpu(), torch.stack(inputs).sum(dim=0)
)

+ @unittest.skipIf(*typed_gpu_unavailable)
def test_merge_pooled_embeddings_meta(self) -> None:
"""
Test that merge_pooled_embeddings works with meta tensor and
@@ -286,6 +302,7 @@ class MergePooledEmbeddingsTest(unittest.TestCase):

assert output_meta.shape == output_cpu.shape

+ @unittest.skipIf(*typed_gpu_unavailable)
def test_merge_pooled_embeddings_empty_input_tensors(self) -> None:
uncat_size = 2
pooled_embeddings = [
diff --git a/fbgemm_gpu/test/sparse/block_bucketize_test.py b/fbgemm_gpu/test/sparse/block_bucketize_test.py
index e7c9837bf..f7857d68b 100644
--- a/fbgemm_gpu/test/sparse/block_bucketize_test.py
Expand Down
1 change: 1 addition & 0 deletions packages/fbgemm-xpu/src/fbgemm_xpu/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
# integrated.
# --------------------------------------------------------------------------
set(host_sources
${CMAKE_CURRENT_SOURCE_DIR}/all_to_one_device.cpp
${CMAKE_CURRENT_SOURCE_DIR}/ops_registry.cpp
${CMAKE_CURRENT_SOURCE_DIR}/sparse_async_cumsum.cpp
${CMAKE_CURRENT_SOURCE_DIR}/split_embeddings_utils_xpu.cpp
Expand Down
68 changes: 68 additions & 0 deletions packages/fbgemm-xpu/src/fbgemm_xpu/all_to_one_device.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
/*
* Copyright (c) Meta Platforms, Inc. and affiliates. All rights reserved.
* Copyright (c) 2026 Intel Corporation. All Rights Reserved.
* SPDX-License-Identifier: BSD-3-Clause
*/

#include <ATen/core/Tensor.h>
#include <ATen/xpu/PeerToPeerAccess.h>
#include <c10/core/DeviceGuard.h>
#include <torch/library.h>

#include <vector>

#include "fbgemm_utils/tensor_utils.h"

namespace fbgemm_xpu {
namespace {

// XPU counterpart of fbgemm_gpu::all_to_one_device (see fbgemm_gpu/src/
// merge_pooled_embedding_ops/merge_pooled_embedding_ops_gpu.cpp).
// It keeps the CUDA contract: the target needs a device index, tensors already
// on the target are returned as they are, and copies are contiguous. CUDA
// copies the rest through its own peer-to-peer path and requires peer access
// between all devices; here Tensor.to lets PyTorch's XPU cross-device copy
// order the source and target streams, and devices without peer access fall
// back to a synchronous copy.
std::vector<at::Tensor> all_to_one_device_xpu(
std::vector<at::Tensor> input_tensors,
at::Device target_device) {
TORCH_CHECK(
target_device.is_xpu(), "all_to_one_device: target_device must be XPU");
TORCH_CHECK(
target_device.has_index(),
"target_device.index() is -1. Please pass target_device with device "
"index, e.g., torch.device(\"xpu:0\")");
for (const auto& tensor : input_tensors) {
TENSOR_ON_SYCL_XPU(tensor);
}

const c10::DeviceGuard guard(target_device);
std::vector<at::Tensor> output_tensors;
output_tensors.reserve(input_tensors.size());
for (const auto& tensor : input_tensors) {
if (tensor.device() == target_device) {
output_tensors.push_back(tensor);
continue;
}
// Without peer access PyTorch stages the copy through pageable host memory
// and does not wait for the device-to-host half when non_blocking is set.
const bool non_blocking = at::xpu::get_p2p_access(
tensor.device().index(), target_device.index());
output_tensors.push_back(tensor.to(
target_device,
tensor.scalar_type(),
non_blocking,
/*copy=*/false,
at::MemoryFormat::Contiguous));
}
return output_tensors;
}

} // namespace

TORCH_LIBRARY_IMPL(fbgemm, XPU, m) {
m.impl("all_to_one_device", TORCH_FN(all_to_one_device_xpu));
}

} // namespace fbgemm_xpu
3 changes: 3 additions & 0 deletions packages/fbgemm-xpu/src/fbgemm_xpu/ops_registry.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,9 @@ PyObject* PyInit__C(void) {
* respective .cpp / .cu files.
*/
TORCH_LIBRARY_FRAGMENT(fbgemm, m) {
if (!utils::torch::schemaExists("fbgemm::all_to_one_device")) {
m.def("all_to_one_device(Tensor[] input_tensors, Device target_device) -> Tensor[]");
}
if (!utils::torch::schemaExists(
"fbgemm::int_nbit_split_embedding_codegen_lookup_function")) {
m.def(
Expand Down
105 changes: 105 additions & 0 deletions packages/fbgemm-xpu/tests/test_all_to_one_device.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
# Copyright (c) 2026 Intel Corporation. All Rights Reserved.
# SPDX-License-Identifier: BSD-3-Clause

"""``all_to_one_device`` on XPU.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are there any tests we can reuse from fbgemm?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, FBGEMM has a test for it.

Addressed in 993e8bb with a hybrid approach: we now run the upstream test on XPU through the shared patch and dropped our copy of it, and keep a few local tests for what upstream doesn't check (no copy for tensors already on the target, contiguous copies, invalid devices, non-default streams).

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you. One thing I really care about is reusing fbgemm tests as much as possible and keeping our own tests to the bare minimum. This way we increase our chances to be really compatible with the upstream and minimize our debt for potential upstreaming.


TorchRec's ``SeqEmbeddingsAllToOne`` gathers per-rank sequence embedding
outputs onto one device with this operator. FBGEMM's patched
``test_all_to_one_device`` checks the gathered values for inputs spread over
devices. These tests cover what it does not check: tensors already on the
target come back as the same views, copies are contiguous, invalid devices are
rejected, and copies work on non-default streams. The cross-device tests need
two XPU devices.
"""

import fbgemm_xpu # noqa: F401 - registers the fbgemm XPU operators
import pytest
import torch

pytestmark = pytest.mark.skipif(
not torch.xpu.is_available(), reason="requires an XPU device"
)

requires_two_xpus = pytest.mark.skipif(
torch.xpu.is_available() and torch.xpu.device_count() < 2,
reason="requires two XPU devices",
)


def test_same_device_preserves_views_and_empty_tensors():
xpu = torch.device("xpu:0")
torch.xpu.set_device(xpu)
base = torch.arange(24, device=xpu).reshape(4, 6)
tensors = [
base,
base[:, ::2],
torch.empty((0, 5), device=xpu),
torch.ones((), device=xpu),
]
outputs = torch.ops.fbgemm.all_to_one_device(tensors, xpu)
assert len(outputs) == len(tensors) # nosec B101
for actual, expected in zip(outputs, tensors):
torch.testing.assert_close(actual, expected)
assert actual.data_ptr() == expected.data_ptr() # nosec B101
assert actual.stride() == expected.stride() # nosec B101


def test_rejects_unsupported_device_inputs():
xpu = torch.device("xpu:0")
tensor = torch.ones(2, device=xpu)
with pytest.raises(RuntimeError, match="target_device must be XPU"):
torch.ops.fbgemm.all_to_one_device([tensor], torch.device("cpu"))
with pytest.raises(RuntimeError, match="Please pass target_device with device index"):
torch.ops.fbgemm.all_to_one_device([tensor], torch.device("xpu"))
with pytest.raises(RuntimeError, match="must be a SYCL XPU tensor"):
torch.ops.fbgemm.all_to_one_device([tensor, torch.ones(2)], xpu)


@requires_two_xpus
def test_mixed_devices_keep_target_views_and_copy_contiguous():
"""In one call, a view already on the target comes back as the same view,
and copies are contiguous whatever the input layout, as on CUDA."""
source = torch.device("xpu:1")
target = torch.device("xpu:0")
on_target = torch.randn((10, 64), device=target)[:, :20]
copied = [
torch.arange(24, device=source).reshape(4, 6).t(),
torch.randn((2, 3, 4, 5), device=source).to(memory_format=torch.channels_last),
torch.randn((10, 64), device=source)[:, :20],
]
outputs = torch.ops.fbgemm.all_to_one_device([on_target, *copied], target)
assert outputs[0].data_ptr() == on_target.data_ptr() # nosec B101
assert outputs[0].stride() == on_target.stride() # nosec B101
for source_tensor, actual in zip(copied, outputs[1:]):
assert actual.device == target # nosec B101
assert actual.is_contiguous() # nosec B101
torch.testing.assert_close(actual.cpu(), source_tensor.cpu())


@requires_two_xpus
def test_cross_device_stream_copy():
"""Copies in both directions on non-default streams."""
for source_index, target_index in ((0, 1), (1, 0)):
source = torch.device(f"xpu:{source_index}")
target = torch.device(f"xpu:{target_index}")
producer = torch.xpu.Stream(device=source)
consumer = torch.xpu.Stream(device=target)
with torch.xpu.stream(producer):
inputs = [
torch.arange(48, device=source).reshape(6, 8)[:, ::2],
torch.full((3, 7), 2.5, dtype=torch.float16, device=source),
torch.empty((0, 4), device=source),
]
torch.xpu.current_stream(source).wait_stream(producer)
with torch.xpu.stream(consumer):
outputs = torch.ops.fbgemm.all_to_one_device(inputs, target)
consumed = [tensor.clone() for tensor in outputs]
consumer.synchronize()
expected = [
torch.arange(48).reshape(6, 8)[:, ::2],
torch.full((3, 7), 2.5, dtype=torch.float16),
torch.empty((0, 4)),
]
for actual, reference in zip(consumed, expected):
assert actual.device == target # nosec B101
torch.testing.assert_close(actual.cpu(), reference)
2 changes: 1 addition & 1 deletion packages/fbgemm-xpu/tests/test_int_nbit_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ def fbgemm():
@pytest.fixture
def xpu(fbgemm):
importlib.import_module("fbgemm_xpu")
assert torch.__version__.split("+")[0] == "2.14.0", torch.__version__ # nosec B101
assert torch.__version__.split("+")[0].startswith("2.14."), torch.__version__ # nosec B101
assert torch.xpu.is_available(), "XPU validation requires a real device; no skip/fallback" # nosec B101
for operator in (LOOKUP, BOUNDS):
assert torch._C._dispatch_has_kernel_for_dispatch_key(operator, "XPU"), operator # nosec B101
Expand Down
Loading