Skip to content
Open
Changes from 2 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
84 changes: 74 additions & 10 deletions cpp/include/raft/linalg/detail/reduce_rows_by_key.cuh
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,10 @@
#include <cub/block/block_reduce.cuh>
#include <cub/util_ptx.cuh>

#include <algorithm>
#include <cstdint>
#include <limits>
#include <type_traits>

#define MAX_BLOCKS 65535u
namespace raft {
Expand Down Expand Up @@ -276,18 +279,22 @@ void sum_rows_by_key_large_nkeys_colmajor(const DataIteratorT d_A,
d_sums);
}

/**
* Large-nkeys reduction with one thread per element, used for rows narrower than a warp
* (ncols < WarpSize), where a warp per row would leave lanes idle.
*/
template <typename DataIteratorT,
typename KeysIteratorT,
typename WeightT,
typename SumsT,
typename IdxT>
RAFT_KERNEL sum_rows_by_key_large_nkeys_kernel_rowmajor(const DataIteratorT d_A,
IdxT lda,
const WeightT* d_weights,
KeysIteratorT d_keys,
IdxT nrows,
IdxT ncols,
SumsT* d_sums)
RAFT_KERNEL sum_rows_by_key_large_nkeys_kernel_rowmajor_per_element(const DataIteratorT d_A,
IdxT lda,
const WeightT* d_weights,
KeysIteratorT d_keys,
IdxT nrows,
IdxT ncols,
SumsT* d_sums)
{
IdxT gid = threadIdx.x + (blockDim.x * static_cast<IdxT>(blockIdx.x));
IdxT j = gid % ncols;
Expand All @@ -299,6 +306,44 @@ RAFT_KERNEL sum_rows_by_key_large_nkeys_kernel_rowmajor(const DataIteratorT d_A,
raft::myAtomicAdd(&d_sums[j + ncols * l], val);
}

/**
* Large-nkeys reduction with one warp per row: the key and weight are loaded once per row and the
* warp's lanes add consecutive columns of the row into d_sums[key].
*/
template <typename DataIteratorT,
typename KeysIteratorT,
typename WeightT,
typename SumsT,

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.

Should we add a static check that SumsT does not narrow WeightT and that both types belong to the same numeric type family?

This is to make sure casts like this don't cause accidental side-effects https://github.com/NVIDIA/raft/pull/3176/changes#diff-ee891f4e924cc29da43803011dfcae91ee2bcec531ba7a447cd0fa2ea5da6e01R376

@huuanhhuyn huuanhhuyn Oct 22, 2026 •

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.

I also don't see any circumstance where we want give weight type more precise or even of a different family type than the sum. I can add it.

@huuanhhuyn huuanhhuyn Oct 27, 2026 •

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.

Added it cf53129.

I assume we still want to allow integer weight type with float sum type. Besides it, integers of different sizes should be fine as they don't cause precision issue.
-> So we throws when both are floats and SumsT does not narrow WeightT.

typename IdxT>
RAFT_KERNEL sum_rows_by_key_large_nkeys_kernel_rowmajor_per_warp(const DataIteratorT d_A,
IdxT lda,
const WeightT* d_weights,
KeysIteratorT d_keys,
IdxT nrows,
IdxT ncols,
SumsT* d_sums)
{
// Plain pointers are read through the read-only cache; other iterators (e.g. a transform
// iterator converting the input type) are dereferenced as is.
constexpr bool kIsPlainPointer =
std::is_same_v<DataIteratorT, const SumsT*> || std::is_same_v<DataIteratorT, SumsT*>;
const IdxT warp_id = (static_cast<IdxT>(blockIdx.x) * blockDim.x + threadIdx.x) / WarpSize;
const IdxT n_warps = static_cast<IdxT>(gridDim.x) * blockDim.x / WarpSize;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
for (IdxT i = warp_id; i < nrows; i += n_warps) {
const auto key = static_cast<IdxT>(d_keys[i]);
const SumsT w = d_weights != nullptr ? static_cast<SumsT>(d_weights[i]) : SumsT(1);
for (IdxT j = laneId(); j < ncols; j += WarpSize) {
SumsT a;
if constexpr (kIsPlainPointer) {
a = __ldg(d_A + lda * i + j);
} else {
a = static_cast<SumsT>(d_A[lda * i + j]);
}
raft::myAtomicAdd(d_sums + ncols * key + j, a * w);
}
}
}

template <typename DataIteratorT,
typename KeysIteratorT,
typename WeightT,
Expand All @@ -313,12 +358,31 @@ void sum_rows_by_key_large_nkeys_rowmajor(const DataIteratorT d_A,
SumsT* d_sums,
cudaStream_t st)
{
uint32_t block_dim = 128;
auto grid_dim = static_cast<uint32_t>(ceildiv<IdxT>(nrows * ncols, (IdxT)block_dim));
constexpr uint32_t block_dim = 128;
if (ncols < WarpSize) {
// Rows narrower than a warp: one thread per element keeps every lane busy.
const auto grid_dim = static_cast<uint32_t>(ceildiv<IdxT>(nrows * ncols, IdxT{block_dim}));
raft::launch_kernel(st,
grid_dim,
block_dim,
sum_rows_by_key_large_nkeys_kernel_rowmajor_per_element,
d_A,
lda,
d_weights,
d_keys,
nrows,
ncols,
d_sums);
return;
}
constexpr uint32_t warps_per_block = block_dim / WarpSize;
// Keep gridDim.x within the launch limit; the kernel's grid-stride loop covers extra rows.
const auto grid_dim =
static_cast<uint32_t>(std::min<IdxT>(ceildiv<IdxT>(nrows, warps_per_block), IdxT{1} << 30));
raft::launch_kernel(st,
grid_dim,
block_dim,
sum_rows_by_key_large_nkeys_kernel_rowmajor,
sum_rows_by_key_large_nkeys_kernel_rowmajor_per_warp,
d_A,
lda,
d_weights,
Expand Down
Loading