Files
wehub-resource-sync 94057c3d3e
PR Test (NPU) / check-changes (push) Has been cancelled
PR Test (NPU) / pr-gate (push) Has been cancelled
PR Test (NPU) / set-image-config (push) Has been cancelled
PR Test (NPU) / stage-b-test-1-npu-a2 (0) (push) Has been cancelled
PR Test (NPU) / stage-b-test-1-npu-a2 (1) (push) Has been cancelled
PR Test (NPU) / stage-b-test-2-npu-a2 (0) (push) Has been cancelled
PR Test (NPU) / stage-b-test-2-npu-a2 (1) (push) Has been cancelled
PR Test (NPU) / stage-b-test-4-npu-a3 (push) Has been cancelled
PR Test (NPU) / stage-b-test-16-npu-a3 (push) Has been cancelled
PR Test (NPU) / multimodal-gen-test-1-npu-a3 (push) Has been cancelled
PR Test (NPU) / multimodal-gen-test-2-npu-a3 (push) Has been cancelled
PR Test (Arm64) / pr-gate (push) Has been cancelled
PR Test (Arm64) / check-changes (push) Has been cancelled
PR Test (Arm64) / build-test (push) Has been cancelled
PR Test (sgl-router) / gate (push) Has been cancelled
PR Test (sgl-router) / tier-1 — lint (push) Has been cancelled
PR Test (sgl-router) / tier-2 — build + test (push) Has been cancelled
PR Test (sgl-router) / tier-3 — docker (placeholder) (push) Has been cancelled
PR Test (sgl-router) / tier-3 — k8s integration (push) Has been cancelled
PR Test (sgl-router) / tier-3 — e2e (push) Has been cancelled
PR Test (sgl-router) / finish (push) Has been cancelled
PR Test (NPU) / single-node-poc (map[name:qwen3_6_27b_w8a8_1p_in64k_out1k_50ms runner:linux-aarch64-a3-2 test_case:test/registered/ascend/performance/qwen3_6_27b/test_npu_qwen3_6_27b_w8a8_1p_in64k_out1k_50ms.py test_type:perf]) (push) Has been cancelled
PR Test (NPU) / pr-test-npu-finish (push) Has been cancelled
PR Test (Xeon) / pr-gate (push) Has been cancelled
PR Test (Xeon) / check-changes (push) Has been cancelled
PR Test (Xeon) / build-test (, xeon-gnr, base-b-test-cpu) (push) Has been cancelled
PR Test (XPU) / check-changes (push) Has been cancelled
PR Test (XPU) / pr-gate (push) Has been cancelled
PR Test (XPU) / stage-a-test-1-gpu-xpu (push) Has been cancelled
PR Test (XPU) / wait-for-stage-a (push) Has been cancelled
PR Test (XPU) / stage-b-test-1-gpu-xpu (push) Has been cancelled
PR Test (XPU) / finish (push) Has been cancelled
CI Model Inventory / build-inventory (push) Has been cancelled
Lint / lint (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark Compilation Check (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark - Manual Policy (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark - Request Processing (push) Has been cancelled
PR Benchmark (SMG Components) / Benchmark Summary (push) Has been cancelled
PR Test (SMG) / build-wheel (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on windows (x86_64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on macos (x86_64 - auto) (push) Has been cancelled
PR Test (SMG) / python-unit-tests (push) Has been cancelled
PR Test (SMG) / unit-tests (push) Has been cancelled
PR Test (SMG) / benchmarks (push) Has been cancelled
PR Test (SMG) / chat-completions (push) Has been cancelled
PR Test (SMG) / chat-completions-4gpu (push) Has been cancelled
PR Test (SMG) / e2e (push) Has been cancelled
PR Test (SMG) / docker-build-test (push) Has been cancelled
PR Test (SMG) / k8s-integration (push) Has been cancelled
PR Test (SMG) / finish (push) Has been cancelled
PR Test (SMG) / summarize-benchmarks (push) Has been cancelled
Release SGLang Model Gateway Docker Image / publish (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on macos (aarch64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (aarch64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (x86_64 - auto) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (aarch64 - musllinux_1_1) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / build on linux (x86_64 - musllinux_1_1) (push) Has been cancelled
Release SGLang Model Gateway to PyPI / Build SDist (push) Has been cancelled
Release SGLang Model Gateway to PyPI / Upload to PyPI (push) Has been cancelled
Release SGLang Kernels / build-cu129-matrix (aarch64, 12.9, 3.10, arm-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / build-cu129-matrix (x86_64, 12.9, 3.10, x64-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / release-cu129 (push) Has been cancelled
Release SGLang Kernels / build-cu130-matrix (aarch64, 13.0, 3.10, arm-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / build-cu130-matrix (x86_64, 13.0, 3.10, x64-kernel-build-node) (push) Has been cancelled
Release SGLang Kernels / release-cu130 (push) Has been cancelled
Release SGLang Kernels / build-rocm-matrix (3.10, 700) (push) Has been cancelled
Release SGLang Kernels / build-rocm-matrix (3.10, 720) (push) Has been cancelled
Release SGLang Kernels / release-rocm700 (push) Has been cancelled
Release SGLang Kernels / release-rocm720 (push) Has been cancelled
Release SGLang Kernels / build-musa43 (43, 3.10) (push) Has been cancelled
Release SGLang Kernels / release-musa43 (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:38:16 +08:00

990 lines
32 KiB
Plaintext

#include <ATen/cuda/CUDAContext.h>
#include <c10/cuda/CUDAException.h>
#include <c10/util/irange.h>
#include <cuda_runtime.h>
#include <cstdint>
#include <limits>
#include <vector>
#if !defined(USE_ROCM) && !defined(USE_MUSA)
#include <dlfcn.h>
#define WARP_SIZE 32
#include "pytorch_extension_utils.h"
#else
#include "pytorch_extension_utils_rocm.h"
#include "utils.h" // WARP_SIZE
#endif
__device__ __forceinline__ void
transfer_item_warp(int32_t lane_id, const void* src_addr, void* dst_addr, int64_t item_size_bytes) {
const uint64_t* __restrict__ src = static_cast<const uint64_t*>(src_addr);
uint64_t* __restrict__ dst = static_cast<uint64_t*>(dst_addr);
const int total_chunks = item_size_bytes / sizeof(uint64_t);
#pragma unroll
for (int j = lane_id; j < total_chunks; j += WARP_SIZE) {
#if !defined(USE_ROCM) && !defined(USE_MUSA)
uint64_t tmp;
asm volatile("ld.global.nc.b64 %0,[%1];" : "=l"(tmp) : "l"(src + j) : "memory");
asm volatile("st.global.cg.b64 [%0],%1;" ::"l"(dst + j), "l"(tmp) : "memory");
#else
uint64_t tmp = __builtin_nontemporal_load(src + j);
__builtin_nontemporal_store(tmp, dst + j);
#endif
}
}
template <typename T>
__device__ __forceinline__ T* get_global_offset_lf(
T* base,
const uintptr_t* __restrict__ /*unused*/,
int64_t layer_id,
int64_t layer_dim,
int64_t page_id,
int64_t item_size_bytes) {
// layer first
return base + layer_id * layer_dim + page_id * item_size_bytes;
}
template <typename T>
__device__ __forceinline__ T* get_global_offset_pf(
T* base,
const uintptr_t* __restrict__ /*unused*/,
int64_t layer_id,
int64_t page_dim,
int64_t page_id,
int64_t item_size_bytes) {
// page first
return base + page_id * page_dim + layer_id * item_size_bytes;
}
// get offset from layer base table when layers are not contiguous
template <typename T>
__device__ __forceinline__ T* get_global_offset_lf_tbl(
T* /*unused*/,
const uintptr_t* __restrict__ layer_base_tbl,
int64_t layer_id,
int64_t /*unused*/,
int64_t page_id,
int64_t item_size_bytes) {
return reinterpret_cast<T*>(layer_base_tbl[layer_id]) + page_id * item_size_bytes;
}
template <typename T>
__device__ __forceinline__ T* get_global_offset_per_head_lf(
T* base,
const uintptr_t* __restrict__ /*unused*/,
int64_t layer_id,
int64_t layer_dim,
int64_t page_id,
int64_t item_size_bytes,
int64_t head_id,
int64_t head_num,
int64_t /*unused*/) {
// layer first offset func per head
return base + layer_id * layer_dim + page_id * item_size_bytes + item_size_bytes / head_num * head_id;
}
template <typename T>
__device__ __forceinline__ T* get_global_offset_per_head_lf_tbl(
T* /*unused*/,
const uintptr_t* __restrict__ layer_base_tbl,
int64_t layer_id,
int64_t /*unused*/,
int64_t page_id,
int64_t item_size_bytes,
int64_t head_id,
int64_t head_num,
int64_t /*unused*/) {
return reinterpret_cast<T*>(layer_base_tbl[layer_id]) + page_id * item_size_bytes +
item_size_bytes / head_num * head_id;
}
template <typename T>
__device__ __forceinline__ T* get_global_offset_ph(
T* base,
const uintptr_t* __restrict__ /*unused*/,
int64_t layer_id,
int64_t page_dim,
int64_t page_id,
int64_t item_size_bytes,
int64_t head_id,
int64_t head_num,
int64_t page_size) {
// page head layout: [page_num, head_num, page_size, layer_num, head_dim]
return base + page_id / page_size * page_size * page_dim + // page_num dimension offset
page_dim / head_num * head_id * page_size + // head_num dimension offset
page_id % page_size * page_dim / head_num + // page_size dimension offset
layer_id * item_size_bytes / head_num; // layer_num dimension offset
}
template <auto SrcOffsetFn, auto DstOffsetFn>
__global__ void transfer_page_head_kernel_impl(
const void* __restrict__ src_k,
void* __restrict__ dst_k,
const void* __restrict__ src_v,
void* __restrict__ dst_v,
const int64_t* __restrict__ src_indices,
const int64_t* __restrict__ dst_indices,
int64_t start_layer_id,
int64_t num_layers_to_process,
int64_t num_items,
int64_t items_per_warp,
int64_t item_size_bytes,
int64_t src_layout_dim,
int64_t dst_layout_dim,
const uintptr_t* __restrict__ src_k_layer_tbl,
const uintptr_t* __restrict__ dst_k_layer_tbl,
const uintptr_t* __restrict__ src_v_layer_tbl,
const uintptr_t* __restrict__ dst_v_layer_tbl,
const int64_t page_size,
const int64_t head_num) {
int32_t tid = blockIdx.x * blockDim.x + threadIdx.x;
int32_t lane_id = tid % WARP_SIZE;
int32_t warp_id = tid / WARP_SIZE;
const int64_t head_size_bytes = item_size_bytes / head_num;
for (int i = 0; i < items_per_warp; ++i) {
int64_t item_id = warp_id * items_per_warp + i;
if (item_id >= num_items) {
break;
}
const int64_t src_page_id = src_indices[item_id];
const int64_t dst_page_id = dst_indices[item_id];
// Loop over layers if necessary
for (int64_t layer_id = start_layer_id; layer_id < start_layer_id + num_layers_to_process; ++layer_id) {
// For page head layout, the cache of each head in the token is discontinuous, need to loop
for (int64_t head_id = 0; head_id < head_num; ++head_id) {
const char* src_k_ptr = SrcOffsetFn(
static_cast<const char*>(src_k),
src_k_layer_tbl,
layer_id,
src_layout_dim,
src_page_id,
item_size_bytes,
head_id,
head_num,
page_size);
char* dst_k_ptr = DstOffsetFn(
static_cast<char*>(dst_k),
dst_k_layer_tbl,
layer_id,
dst_layout_dim,
dst_page_id,
item_size_bytes,
head_id,
head_num,
page_size);
transfer_item_warp(lane_id, src_k_ptr, dst_k_ptr, head_size_bytes);
const char* src_v_ptr = SrcOffsetFn(
static_cast<const char*>(src_v),
src_v_layer_tbl,
layer_id,
src_layout_dim,
src_page_id,
item_size_bytes,
head_id,
head_num,
page_size);
char* dst_v_ptr = DstOffsetFn(
static_cast<char*>(dst_v),
dst_v_layer_tbl,
layer_id,
dst_layout_dim,
dst_page_id,
item_size_bytes,
head_id,
head_num,
page_size);
transfer_item_warp(lane_id, src_v_ptr, dst_v_ptr, head_size_bytes);
}
}
}
}
template <auto SrcOffsetFn, auto DstOffsetFn, bool IsMLA>
__global__ void transfer_kernel_impl(
const void* __restrict__ src_k,
void* __restrict__ dst_k,
const void* __restrict__ src_v,
void* __restrict__ dst_v,
const int64_t* __restrict__ src_indices,
const int64_t* __restrict__ dst_indices,
int64_t start_layer_id,
int64_t num_layers_to_process,
int64_t num_items,
int64_t items_per_warp,
int64_t item_size_bytes,
int64_t src_layout_dim,
int64_t dst_layout_dim,
const uintptr_t* __restrict__ src_k_layer_tbl,
const uintptr_t* __restrict__ dst_k_layer_tbl,
const uintptr_t* __restrict__ src_v_layer_tbl,
const uintptr_t* __restrict__ dst_v_layer_tbl) {
int32_t tid = blockIdx.x * blockDim.x + threadIdx.x;
int32_t lane_id = tid % WARP_SIZE;
int32_t warp_id = tid / WARP_SIZE;
for (int i = 0; i < items_per_warp; ++i) {
int64_t item_id = warp_id * items_per_warp + i;
if (item_id >= num_items) {
break;
}
const int64_t src_page_id = src_indices[item_id];
const int64_t dst_page_id = dst_indices[item_id];
// Loop over layers if necessary
for (int64_t layer_id = start_layer_id; layer_id < start_layer_id + num_layers_to_process; ++layer_id) {
const char* src_ptr = SrcOffsetFn(
static_cast<const char*>(src_k), src_k_layer_tbl, layer_id, src_layout_dim, src_page_id, item_size_bytes);
char* dst_ptr = DstOffsetFn(
static_cast<char*>(dst_k), dst_k_layer_tbl, layer_id, dst_layout_dim, dst_page_id, item_size_bytes);
transfer_item_warp(lane_id, src_ptr, dst_ptr, item_size_bytes);
if constexpr (!IsMLA) {
const char* src_v_ptr = SrcOffsetFn(
static_cast<const char*>(src_v), src_v_layer_tbl, layer_id, src_layout_dim, src_page_id, item_size_bytes);
char* dst_v_ptr = DstOffsetFn(
static_cast<char*>(dst_v), dst_v_layer_tbl, layer_id, dst_layout_dim, dst_page_id, item_size_bytes);
transfer_item_warp(lane_id, src_v_ptr, dst_v_ptr, item_size_bytes);
}
}
}
}
template <auto SrcOffsetFn, auto DstOffsetFn, bool IsMLA, bool PageHeadLayout = false>
void transfer_kv_launcher(
const at::Tensor& src_k,
at::Tensor& dst_k,
const at::Tensor& src_v,
at::Tensor& dst_v,
const at::Tensor& src_indices,
const at::Tensor& dst_indices,
int64_t start_layer_id,
int64_t num_layers_to_process,
int64_t item_size,
int64_t src_layout_dim,
int64_t dst_layout_dim,
const at::Tensor& src_k_layers,
const at::Tensor& dst_k_layers,
const at::Tensor& src_v_layers,
const at::Tensor& dst_v_layers,
int64_t block_quota,
int64_t num_warps_per_block,
const int64_t page_size = 16,
const int64_t head_num = 1) {
TORCH_CHECK(src_indices.is_cuda(), "Source indices must be a CUDA tensor");
TORCH_CHECK(dst_indices.is_cuda(), "Destination indices must be a CUDA tensor");
TORCH_CHECK(src_indices.scalar_type() == at::kLong, "Source indices must be of type long");
TORCH_CHECK(dst_indices.scalar_type() == at::kLong, "Destination indices must be of type long");
TORCH_CHECK(src_indices.numel() == dst_indices.numel(), "Source and destination indices must have the same length");
TORCH_CHECK(item_size % 8 == 0, "Item byte size must be divisible by 8");
auto div_up = [](int64_t x, int64_t y) { return (x + y - 1) / y; };
const int64_t num_items = src_indices.numel();
const int64_t items_per_warp = div_up(num_items, block_quota * num_warps_per_block);
const int32_t num_blocks = div_up(num_items, items_per_warp * num_warps_per_block);
dim3 grid_dim(num_blocks, 1, 1);
const int32_t threads_per_block = num_warps_per_block * WARP_SIZE;
const void* src_k_ptr = src_k.defined() ? src_k.data_ptr() : nullptr;
void* dst_k_ptr = dst_k.defined() ? dst_k.data_ptr() : nullptr;
const void* src_v_ptr = IsMLA || !src_v.defined() ? nullptr : src_v.data_ptr();
void* dst_v_ptr = IsMLA || !dst_v.defined() ? nullptr : dst_v.data_ptr();
const uintptr_t* src_k_tbl_ptr = src_k_layers.defined() ? src_k_layers.data_ptr<uintptr_t>() : nullptr;
const uintptr_t* dst_k_tbl_ptr = dst_k_layers.defined() ? dst_k_layers.data_ptr<uintptr_t>() : nullptr;
const uintptr_t* src_v_tbl_ptr = IsMLA || !src_v_layers.defined() ? nullptr : src_v_layers.data_ptr<uintptr_t>();
const uintptr_t* dst_v_tbl_ptr = IsMLA || !dst_v_layers.defined() ? nullptr : dst_v_layers.data_ptr<uintptr_t>();
cudaStream_t torch_current_stream = at::cuda::getCurrentCUDAStream();
if constexpr (PageHeadLayout) {
transfer_page_head_kernel_impl<SrcOffsetFn, DstOffsetFn><<<grid_dim, threads_per_block, 0, torch_current_stream>>>(
src_k_ptr,
dst_k_ptr,
src_v_ptr,
dst_v_ptr,
src_indices.data_ptr<int64_t>(),
dst_indices.data_ptr<int64_t>(),
start_layer_id,
num_layers_to_process,
num_items,
items_per_warp,
item_size,
src_layout_dim,
dst_layout_dim,
src_k_tbl_ptr,
dst_k_tbl_ptr,
src_v_tbl_ptr,
dst_v_tbl_ptr,
page_size,
head_num);
} else {
transfer_kernel_impl<SrcOffsetFn, DstOffsetFn, IsMLA><<<grid_dim, threads_per_block, 0, torch_current_stream>>>(
src_k_ptr,
dst_k_ptr,
src_v_ptr,
dst_v_ptr,
src_indices.data_ptr<int64_t>(),
dst_indices.data_ptr<int64_t>(),
start_layer_id,
num_layers_to_process,
num_items,
items_per_warp,
item_size,
src_layout_dim,
dst_layout_dim,
src_k_tbl_ptr,
dst_k_tbl_ptr,
src_v_tbl_ptr,
dst_v_tbl_ptr);
}
C10_CUDA_KERNEL_LAUNCH_CHECK();
}
void transfer_kv_per_layer(
const at::Tensor src_k,
at::Tensor dst_k,
const at::Tensor src_v,
at::Tensor dst_v,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t block_quota,
int64_t num_warps_per_block) {
at::Tensor empty;
transfer_kv_launcher<get_global_offset_lf<const char>, get_global_offset_lf<char>, false>(
src_k,
dst_k,
src_v,
dst_v,
src_indices,
dst_indices,
0,
1,
item_size,
0,
0,
empty,
empty,
empty,
empty,
block_quota,
num_warps_per_block);
}
void transfer_kv_per_layer_pf_lf(
const at::Tensor src_k,
at::Tensor dst_k,
const at::Tensor src_v,
at::Tensor dst_v,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t layer_id,
int64_t item_size,
int64_t src_layout_dim,
int64_t block_quota,
int64_t num_warps_per_block) {
at::Tensor empty;
transfer_kv_launcher<get_global_offset_pf<const char>, get_global_offset_lf<char>, false>(
src_k,
dst_k,
src_v,
dst_v,
src_indices,
dst_indices,
layer_id,
1,
item_size,
src_layout_dim,
0,
empty,
empty,
empty,
empty,
block_quota,
num_warps_per_block);
}
void transfer_kv_per_layer_ph_lf(
const at::Tensor src_k,
at::Tensor dst_k,
const at::Tensor src_v,
at::Tensor dst_v,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t layer_id,
int64_t item_size,
int64_t src_layout_dim,
int64_t page_size,
int64_t head_num,
int64_t block_quota,
int64_t num_warps_per_block) {
at::Tensor empty;
transfer_kv_launcher<get_global_offset_ph<const char>, get_global_offset_per_head_lf<char>, false, true>(
src_k,
dst_k,
src_v,
dst_v,
src_indices,
dst_indices,
layer_id,
1,
item_size,
src_layout_dim,
0,
empty,
empty,
empty,
empty,
block_quota,
num_warps_per_block,
page_size,
head_num);
}
void transfer_kv_all_layer(
const at::Tensor src_k_layers,
const at::Tensor dst_k_layers,
const at::Tensor src_v_layers,
const at::Tensor dst_v_layers,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t num_layers,
int64_t block_quota,
int64_t num_warps_per_block) {
TORCH_CHECK(num_layers == src_k_layers.size(0), "Number of layers in source k tensor does not match num_layers");
at::Tensor empty;
transfer_kv_launcher<get_global_offset_lf_tbl<const char>, get_global_offset_lf_tbl<char>, false>(
empty,
empty,
empty,
empty,
src_indices,
dst_indices,
0,
num_layers,
item_size,
0,
0,
src_k_layers,
dst_k_layers,
src_v_layers,
dst_v_layers,
block_quota,
num_warps_per_block);
}
void transfer_kv_all_layer_lf_pf(
const at::Tensor src_k_layers,
at::Tensor dst_k,
const at::Tensor src_v_layers,
at::Tensor dst_v,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t dst_layout_dim,
int64_t num_layers,
int64_t block_quota,
int64_t num_warps_per_block) {
TORCH_CHECK(num_layers == src_k_layers.size(0), "Number of layers in source k tensor does not match num_layers");
at::Tensor empty;
transfer_kv_launcher<get_global_offset_lf_tbl<const char>, get_global_offset_pf<char>, false>(
empty,
dst_k,
empty,
dst_v,
src_indices,
dst_indices,
0,
num_layers,
item_size,
0,
dst_layout_dim,
src_k_layers,
empty,
src_v_layers,
empty,
block_quota,
num_warps_per_block);
}
void transfer_kv_all_layer_lf_ph(
const at::Tensor src_k_layers,
at::Tensor dst_k,
const at::Tensor src_v_layers,
at::Tensor dst_v,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t dst_layout_dim,
int64_t num_layers,
int64_t page_size,
int64_t head_num,
int64_t block_quota,
int64_t num_warps_per_block) {
TORCH_CHECK(num_layers == src_k_layers.size(0), "Number of layers in source k tensor does not match num_layers");
at::Tensor empty;
transfer_kv_launcher<get_global_offset_per_head_lf_tbl<const char>, get_global_offset_ph<char>, false, true>(
empty,
dst_k,
empty,
dst_v,
src_indices,
dst_indices,
0,
num_layers,
item_size,
0,
dst_layout_dim,
src_k_layers,
empty,
src_v_layers,
empty,
block_quota,
num_warps_per_block,
page_size,
head_num);
}
void transfer_kv_per_layer_mla(
const at::Tensor src,
at::Tensor dst,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t block_quota,
int64_t num_warps_per_block) {
at::Tensor empty;
transfer_kv_launcher<get_global_offset_lf<const char>, get_global_offset_lf<char>, true>(
src,
dst,
empty,
empty,
src_indices,
dst_indices,
0,
1,
item_size,
0,
0,
empty,
empty,
empty,
empty,
block_quota,
num_warps_per_block);
}
void transfer_kv_per_layer_mla_pf_lf(
const at::Tensor src,
at::Tensor dst,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t layer_id,
int64_t item_size,
int64_t src_layout_dim,
int64_t block_quota,
int64_t num_warps_per_block) {
at::Tensor empty;
transfer_kv_launcher<get_global_offset_pf<const char>, get_global_offset_lf<char>, true>(
src,
dst,
empty,
empty,
src_indices,
dst_indices,
layer_id,
1,
item_size,
src_layout_dim,
0,
empty,
empty,
empty,
empty,
block_quota,
num_warps_per_block);
}
void transfer_kv_all_layer_mla(
const at::Tensor src_layers,
const at::Tensor dst_layers,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t num_layers,
int64_t block_quota,
int64_t num_warps_per_block) {
TORCH_CHECK(num_layers == src_layers.size(0), "Number of layers in source tensor does not match num_layers");
at::Tensor empty;
transfer_kv_launcher<get_global_offset_lf_tbl<const char>, get_global_offset_lf_tbl<char>, true>(
empty,
empty,
empty,
empty,
src_indices,
dst_indices,
0,
num_layers,
item_size,
0,
0,
src_layers,
dst_layers,
empty,
empty,
block_quota,
num_warps_per_block);
}
void transfer_kv_all_layer_mla_lf_pf(
const at::Tensor src_layers,
at::Tensor dst,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t item_size,
int64_t dst_layout_dim,
int64_t num_layers,
int64_t block_quota,
int64_t num_warps_per_block) {
TORCH_CHECK(num_layers == src_layers.size(0), "Number of layers in source tensor does not match num_layers");
at::Tensor empty;
transfer_kv_launcher<get_global_offset_lf_tbl<const char>, get_global_offset_pf<char>, true>(
empty,
dst,
empty,
empty,
src_indices,
dst_indices,
0,
num_layers,
item_size,
0,
dst_layout_dim,
src_layers,
empty,
empty,
empty,
block_quota,
num_warps_per_block);
}
inline void transfer_page_direct(
const at::Tensor src_buffer,
at::Tensor dst_buffer,
int64_t src_page_index,
int64_t dst_page_index,
int64_t page_size) {
dst_buffer.slice(0, dst_page_index, dst_page_index + page_size)
.copy_(
src_buffer.slice(0, src_page_index, src_page_index + page_size),
/* non_blocking= */ true);
}
void transfer_kv_direct(
const std::vector<at::Tensor>& src_layers,
std::vector<at::Tensor> dst_layers,
const at::Tensor src_indices,
const at::Tensor dst_indices,
int64_t page_size) {
TORCH_CHECK(
src_layers.size() == dst_layers.size(), "Source and destination layers must have the same number of layers");
TORCH_CHECK(src_indices.numel() == dst_indices.numel(), "Source and destination indices must have the same length");
TORCH_CHECK(page_size > 0, "Page size must be positive");
TORCH_CHECK(src_indices.numel() % page_size == 0, "Source indices size must be divisible by page size");
auto src_indices_cpu = src_indices.cpu();
auto dst_indices_cpu = dst_indices.cpu();
const auto num_indices = src_indices_cpu.numel();
const int64_t num_layers = src_layers.size();
int64_t* src_indices_ptr = src_indices_cpu.data_ptr<int64_t>();
int64_t* dst_indices_ptr = dst_indices_cpu.data_ptr<int64_t>();
int64_t start_index = 0;
int64_t end_index = 0;
for (int64_t i = 0; i < num_indices; ++i) {
if (i < num_indices - 1) {
auto src_diff = src_indices_ptr[i + 1] - src_indices_ptr[i];
auto dst_diff = dst_indices_ptr[i + 1] - dst_indices_ptr[i];
if (src_diff == 1 && dst_diff == 1) {
continue;
}
end_index = i + 1;
} else { // last batch
end_index = num_indices;
}
auto src_index = src_indices_ptr[start_index];
auto dst_index = dst_indices_ptr[start_index];
auto num_tokens = end_index - start_index;
for (int64_t j = 0; j < num_layers; ++j) {
transfer_page_direct(src_layers[j], dst_layers[j], src_index, dst_index, num_tokens);
}
start_index = end_index;
}
}
template <bool IsLf2Pf>
inline void transfer_kv_page_first_direct_impl(
const std::vector<at::Tensor>& src_ptrs,
std::vector<at::Tensor> dst_ptrs,
const at::Tensor& src_indices,
const at::Tensor& dst_indices,
int64_t start_layer_id,
int64_t page_size) {
TORCH_CHECK(src_indices.numel() == dst_indices.numel(), "Source and destination indices must have the same length");
TORCH_CHECK(page_size > 0, "Page size must be positive");
TORCH_CHECK(src_indices.numel() % page_size == 0, "Source indices size must be divisible by page size");
auto src_indices_cpu = src_indices.cpu();
auto dst_indices_cpu = dst_indices.cpu();
const int64_t num_pages = src_indices_cpu.size(0) / page_size;
int64_t* src_indices_ptr = src_indices_cpu.data_ptr<int64_t>();
int64_t* dst_indices_ptr = dst_indices_cpu.data_ptr<int64_t>();
auto fallback_to_page_copy = [&]() {
if constexpr (IsLf2Pf) {
const bool is_mla = dst_ptrs.size() == 1;
const int64_t num_layers = is_mla ? src_ptrs.size() : src_ptrs.size() / 2;
for (const auto i : c10::irange(num_pages)) {
const int64_t s_index = src_indices_ptr[i * page_size];
const int64_t d_index = dst_indices_ptr[i * page_size] / page_size;
for (int64_t j = 0; j < num_layers; ++j) {
transfer_page_direct(
src_ptrs[j], dst_ptrs[0].select(0, d_index).select(0, start_layer_id + j), s_index, 0, page_size);
if (!is_mla) {
transfer_page_direct(
src_ptrs[j + num_layers],
dst_ptrs[1].select(0, d_index).select(0, start_layer_id + j),
s_index,
0,
page_size);
}
}
}
} else {
const bool is_mla = src_ptrs.size() == 1;
const int64_t num_layers = is_mla ? dst_ptrs.size() : dst_ptrs.size() / 2;
for (const auto i : c10::irange(num_pages)) {
const int64_t s_index = src_indices_ptr[i * page_size] / page_size;
const int64_t d_index = dst_indices_ptr[i * page_size];
for (int64_t j = 0; j < num_layers; ++j) {
transfer_page_direct(
src_ptrs[0].select(0, s_index).select(0, start_layer_id + j), dst_ptrs[j], 0, d_index, page_size);
if (!is_mla) {
transfer_page_direct(
src_ptrs[1].select(0, s_index).select(0, start_layer_id + j),
dst_ptrs[j + num_layers],
0,
d_index,
page_size);
}
}
}
}
};
#if defined(USE_ROCM) || !defined(CUDA_VERSION) || CUDA_VERSION < 12080
fallback_to_page_copy();
return;
#else
// Driver capability gate: only use cudaMemcpyBatchAsync on CUDA 12.8+ drivers.
int driver_version = 0;
cudaError_t driver_version_err = cudaDriverGetVersion(&driver_version);
if (driver_version_err != cudaSuccess || driver_version < 12080) {
fallback_to_page_copy();
return;
}
// Symbol gate: runtime may not expose cudaMemcpyBatchAsync in some environments.
static void* cuda_memcpy_batch_async_sym = dlsym(RTLD_DEFAULT, "cudaMemcpyBatchAsync");
if (cuda_memcpy_batch_async_sym == nullptr) {
fallback_to_page_copy();
return;
}
// CUDA 13.0 removed the failIdx parameter from cudaMemcpyBatchAsync. The ABI
// of the dlsym'd symbol is determined by the libcudart loaded in this process,
// not the host driver — a cu12 runtime on a cu13 driver host (common in
// containers) still exposes the 9-param v12 signature. Dispatching on the
// driver version here would segfault in that case (verified empirically).
// Use cudaRuntimeGetVersion so the signature follows the runtime. The
// runtime version is process-constant, so cache the query (static init is
// thread-safe in C++11+) to keep the KV-transfer hot path free of a redundant
// runtime API call per invocation.
static int runtime_version = 0;
static cudaError_t runtime_version_err = cudaRuntimeGetVersion(&runtime_version);
if (runtime_version_err != cudaSuccess) {
fallback_to_page_copy();
return;
}
static const bool use_v13_signature = runtime_version >= 13000;
size_t num_copies = 0;
std::vector<void*> batch_srcs;
std::vector<void*> batch_dsts;
std::vector<size_t> batch_sizes;
std::vector<size_t> attrs_idxs(1, 0);
cudaMemcpyAttributes attrs{};
const int device_id = at::cuda::current_device();
const cudaStream_t stream = at::cuda::getCurrentCUDAStream();
auto append_copy = [&](void* src, void* dst, size_t size_bytes) {
batch_srcs.push_back(src);
batch_dsts.push_back(dst);
batch_sizes.push_back(size_bytes);
};
if constexpr (IsLf2Pf) {
const bool is_mla = dst_ptrs.size() == 1;
const int64_t num_layers = is_mla ? src_ptrs.size() : src_ptrs.size() / 2;
const int64_t dst_stride0 = dst_ptrs[0].stride(0);
const int64_t dst_stride1 = dst_ptrs[0].stride(1);
const int64_t src_stride0 = src_ptrs[0].stride(0);
const int64_t elem_size = dst_ptrs[0].element_size();
const int64_t copy_size_bytes = page_size * src_stride0 * elem_size;
attrs.srcAccessOrder = cudaMemcpySrcAccessOrderStream;
attrs.srcLocHint.type = cudaMemLocationTypeDevice;
attrs.srcLocHint.id = device_id;
attrs.dstLocHint.type = cudaMemLocationTypeHost;
attrs.dstLocHint.id = 0;
attrs.flags = 0;
num_copies = static_cast<size_t>(num_pages) * static_cast<size_t>(num_layers) * static_cast<size_t>(is_mla ? 1 : 2);
batch_srcs.reserve(num_copies);
batch_dsts.reserve(num_copies);
batch_sizes.reserve(num_copies);
for (const auto i : c10::irange(num_pages)) {
auto s_index = src_indices_ptr[i * page_size];
auto d_index = dst_indices_ptr[i * page_size] / page_size;
for (int64_t j = 0; j < num_layers; ++j) {
const char* src_k_ptr = static_cast<const char*>(src_ptrs[j].data_ptr()) + s_index * src_stride0 * elem_size;
char* dst_k_ptr = static_cast<char*>(dst_ptrs[0].data_ptr()) + d_index * dst_stride0 * elem_size +
(start_layer_id + j) * dst_stride1 * elem_size;
append_copy(const_cast<char*>(src_k_ptr), dst_k_ptr, copy_size_bytes);
if (!is_mla) {
const char* src_v_ptr =
static_cast<const char*>(src_ptrs[j + num_layers].data_ptr()) + s_index * src_stride0 * elem_size;
char* dst_v_ptr = static_cast<char*>(dst_ptrs[1].data_ptr()) + d_index * dst_stride0 * elem_size +
(start_layer_id + j) * dst_stride1 * elem_size;
append_copy(const_cast<char*>(src_v_ptr), dst_v_ptr, copy_size_bytes);
}
}
}
} else {
const bool is_mla = src_ptrs.size() == 1;
const int64_t num_layers = is_mla ? dst_ptrs.size() : dst_ptrs.size() / 2;
const int64_t src_stride0 = src_ptrs[0].stride(0);
const int64_t src_stride1 = src_ptrs[0].stride(1);
const int64_t dst_stride0 = dst_ptrs[0].stride(0);
const int64_t elem_size = src_ptrs[0].element_size();
const int64_t copy_size_bytes = page_size * dst_stride0 * elem_size;
attrs.srcAccessOrder = cudaMemcpySrcAccessOrderStream;
attrs.srcLocHint.type = cudaMemLocationTypeHost;
attrs.srcLocHint.id = 0;
attrs.dstLocHint.type = cudaMemLocationTypeDevice;
attrs.dstLocHint.id = device_id;
attrs.flags = 0;
num_copies = static_cast<size_t>(num_pages) * static_cast<size_t>(num_layers) * static_cast<size_t>(is_mla ? 1 : 2);
batch_srcs.reserve(num_copies);
batch_dsts.reserve(num_copies);
batch_sizes.reserve(num_copies);
for (const auto i : c10::irange(num_pages)) {
auto s_index = src_indices_ptr[i * page_size] / page_size;
auto d_index = dst_indices_ptr[i * page_size];
for (int64_t j = 0; j < num_layers; ++j) {
const char* src_k_ptr = static_cast<const char*>(src_ptrs[0].data_ptr()) + s_index * src_stride0 * elem_size +
(start_layer_id + j) * src_stride1 * elem_size;
char* dst_k_ptr = static_cast<char*>(dst_ptrs[j].data_ptr()) + d_index * dst_stride0 * elem_size;
append_copy(const_cast<char*>(src_k_ptr), dst_k_ptr, copy_size_bytes);
if (!is_mla) {
const char* src_v_ptr = static_cast<const char*>(src_ptrs[1].data_ptr()) + s_index * src_stride0 * elem_size +
(start_layer_id + j) * src_stride1 * elem_size;
char* dst_v_ptr = static_cast<char*>(dst_ptrs[j + num_layers].data_ptr()) + d_index * dst_stride0 * elem_size;
append_copy(const_cast<char*>(src_v_ptr), dst_v_ptr, copy_size_bytes);
}
}
}
}
TORCH_CHECK(batch_srcs.size() == num_copies, "Batch memcpy count mismatch");
if (num_copies > 0) {
cudaError_t err;
size_t fail_idx = std::numeric_limits<size_t>::max();
if (use_v13_signature) {
using FnV13 = cudaError_t (*)(
void* const*,
const void* const*,
const size_t*,
size_t,
cudaMemcpyAttributes*,
size_t*,
size_t,
cudaStream_t);
auto fn = reinterpret_cast<FnV13>(cuda_memcpy_batch_async_sym);
err = fn(
batch_dsts.data(), batch_srcs.data(), batch_sizes.data(), num_copies, &attrs, attrs_idxs.data(), 1, stream);
} else {
using FnV12 = cudaError_t (*)(
void**, void**, size_t*, size_t, cudaMemcpyAttributes*, size_t*, size_t, size_t*, cudaStream_t);
auto fn = reinterpret_cast<FnV12>(cuda_memcpy_batch_async_sym);
err =
fn(batch_dsts.data(),
batch_srcs.data(),
batch_sizes.data(),
num_copies,
&attrs,
attrs_idxs.data(),
1,
&fail_idx,
stream);
}
if (err == cudaErrorNotSupported || err == cudaErrorCallRequiresNewerDriver) {
fallback_to_page_copy();
return;
}
if (err != cudaSuccess) {
TORCH_CHECK(false, "cudaMemcpyBatchAsync failed. failIdx=", fail_idx, " error=", cudaGetErrorString(err));
}
}
#endif
}
void transfer_kv_per_layer_direct_pf_lf(
const std::vector<at::Tensor>& src_ptrs,
std::vector<at::Tensor> dst_ptrs,
const at::Tensor& src_indices,
const at::Tensor& dst_indices,
int64_t layer_id,
int64_t page_size) {
transfer_kv_page_first_direct_impl<false>(src_ptrs, dst_ptrs, src_indices, dst_indices, layer_id, page_size);
}
void transfer_kv_all_layer_direct_lf_pf(
const std::vector<at::Tensor>& src_ptrs,
std::vector<at::Tensor> dst_ptrs,
const at::Tensor& src_indices,
const at::Tensor& dst_indices,
int64_t page_size) {
transfer_kv_page_first_direct_impl<true>(src_ptrs, dst_ptrs, src_indices, dst_indices, 0, page_size);
}