minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 13 Packages Projects Releases Wiki Activity

[HiCache] Supports Indexer layout for NSATokenToKVPoolHost (#19912)

Browse Source 
Co-authored-by: hzh0425 <hzh0425@apache.org>
This commit is contained in:
huangtingwei
2026-03-09 18:24:06 -07:00
committed by GitHub
co-authored by hzh0425
parent 0fd9a57d80
commit 254d3cee0b
1 changed files with 113 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/mem_cache/memory_pool_host.py
+113 -47
View File
@@ -1109,6 +1109,7 @@ class NSATokenToKVPoolHost(MLATokenToKVPoolHost):
self.indexer_page_stride_size = ( self.indexer_page_stride_size = (
self.indexer_size_per_token * self.page_size * self.indexer_dtype.itemsize self.indexer_size_per_token * self.page_size * self.indexer_dtype.itemsize
) )
self.indexer_layout_dim = self.indexer_page_stride_size * self.layer_num
self.indexer_page_num = (self.size + self.page_size + 1) // self.page_size self.indexer_page_num = (self.size + self.page_size + 1) // self.page_size
self._init_indexer_buffers() self._init_indexer_buffers()
logger.info( logger.info(
@@ -1124,29 +1125,45 @@ class NSATokenToKVPoolHost(MLATokenToKVPoolHost):
def _init_indexer_buffers(self): def _init_indexer_buffers(self):
alloc_func = ALLOC_MEMORY_FUNCS[self.device_pool.device] alloc_func = ALLOC_MEMORY_FUNCS[self.device_pool.device]
self.index_k_with_scale_buffer = [
alloc_func(
(self.indexer_page_num, self.indexer_page_stride_size),
dtype=self.indexer_dtype,
device=self.device,
pin_memory=self.pin_memory,
allocator=self.allocator,
)
for _ in range(self.layer_num)
]
self.index_k_data_refs = [
self.index_k_with_scale_buffer[i] for i in range(self.layer_num)
]
self.index_k_data_ptrs = torch.tensor(
[x.data_ptr() for x in self.index_k_data_refs],
dtype=torch.uint64,
device=self.device_pool.device,
)
self.index_k_device_ptrs = torch.tensor( self.index_k_device_ptrs = torch.tensor(
[x.data_ptr() for x in self.device_pool.index_k_with_scale_buffer], [x.data_ptr() for x in self.device_pool.index_k_with_scale_buffer],
dtype=torch.uint64, dtype=torch.uint64,
device=self.device_pool.device, device=self.device_pool.device,
) )
if self.layout == "layer_first":
self.index_k_with_scale_buffer = [
alloc_func(
(self.indexer_page_num, self.indexer_page_stride_size),
dtype=self.indexer_dtype,
device=self.device,
pin_memory=self.pin_memory,
allocator=self.allocator,
)
for _ in range(self.layer_num)
]
self.index_k_data_refs = [
self.index_k_with_scale_buffer[i] for i in range(self.layer_num)
]
self.index_k_data_ptrs = torch.tensor(
[x.data_ptr() for x in self.index_k_data_refs],
dtype=torch.uint64,
device=self.device_pool.device,
)
elif self.layout in ["page_first", "page_first_direct"]:
self.index_k_with_scale_buffer = alloc_func(
(
self.indexer_page_num,
self.layer_num,
1,
self.indexer_page_stride_size,
),
dtype=self.indexer_dtype,
device=self.device,
pin_memory=self.pin_memory,
allocator=self.allocator,
)
else:
raise ValueError(f"Unsupported layout: {self.layout}")
def _get_indexer_page_indices(self, host_indices, device_indices): def _get_indexer_page_indices(self, host_indices, device_indices):
if host_indices.numel() == 0: if host_indices.numel() == 0:
@@ -1171,21 +1188,46 @@ class NSATokenToKVPoolHost(MLATokenToKVPoolHost):
) )
use_kernel = io_backend == "kernel" and self.indexer_page_stride_size % 8 == 0 use_kernel = io_backend == "kernel" and self.indexer_page_stride_size % 8 == 0
if use_kernel: if use_kernel:
transfer_kv_per_layer_mla( if self.layout == "layer_first":
src=self.index_k_with_scale_buffer[layer_id], transfer_kv_per_layer_mla(
dst=device_pool.index_k_with_scale_buffer[layer_id], src=self.index_k_with_scale_buffer[layer_id],
src_indices=host_page_indices, dst=device_pool.index_k_with_scale_buffer[layer_id],
dst_indices=device_page_indices, src_indices=host_page_indices,
item_size=self.indexer_page_stride_size, dst_indices=device_page_indices,
) item_size=self.indexer_page_stride_size,
)
elif self.layout == "page_first":
transfer_kv_per_layer_mla_pf_lf(
src=self.index_k_with_scale_buffer,
dst=device_pool.index_k_with_scale_buffer[layer_id],
src_indices=host_page_indices,
dst_indices=device_page_indices,
layer_id=layer_id,
item_size=self.indexer_page_stride_size,
src_layout_dim=self.indexer_layout_dim,
)
else:
raise ValueError(f"Unsupported layout: {self.layout}")
else: else:
transfer_kv_direct( if self.layout == "layer_first":
src_layers=[self.index_k_with_scale_buffer[layer_id]], transfer_kv_direct(
dst_layers=[device_pool.index_k_with_scale_buffer[layer_id]], src_layers=[self.index_k_with_scale_buffer[layer_id]],
src_indices=host_page_indices, dst_layers=[device_pool.index_k_with_scale_buffer[layer_id]],
dst_indices=device_page_indices, src_indices=host_page_indices,
page_size=1, dst_indices=device_page_indices,
) page_size=1,
)
elif self.layout == "page_first_direct":
transfer_kv_per_layer_direct_pf_lf(
src_ptrs=[self.index_k_with_scale_buffer],
dst_ptrs=[device_pool.index_k_with_scale_buffer[layer_id]],
src_indices=host_page_indices,
dst_indices=device_page_indices,
layer_id=layer_id,
page_size=1,
)
else:
raise ValueError(f"Unsupported layout: {self.layout}")
def _backup_indexer_from_device_all_layer( def _backup_indexer_from_device_all_layer(
self, device_pool, host_indices, device_indices, io_backend self, device_pool, host_indices, device_indices, io_backend
@@ -1195,22 +1237,46 @@ class NSATokenToKVPoolHost(MLATokenToKVPoolHost):
) )
use_kernel = io_backend == "kernel" and self.indexer_page_stride_size % 8 == 0 use_kernel = io_backend == "kernel" and self.indexer_page_stride_size % 8 == 0
if use_kernel: if use_kernel:
transfer_kv_all_layer_mla( if self.layout == "layer_first":
src_layers=self.index_k_device_ptrs, transfer_kv_all_layer_mla(
dst_layers=self.index_k_data_ptrs, src_layers=self.index_k_device_ptrs,
src_indices=device_page_indices, dst_layers=self.index_k_data_ptrs,
dst_indices=host_page_indices, src_indices=device_page_indices,
item_size=self.indexer_page_stride_size, dst_indices=host_page_indices,
num_layers=self.layer_num, item_size=self.indexer_page_stride_size,
) num_layers=self.layer_num,
)
elif self.layout == "page_first":
transfer_kv_all_layer_mla_lf_pf(
src_layers=self.index_k_device_ptrs,
dst=self.index_k_with_scale_buffer,
src_indices=device_page_indices,
dst_indices=host_page_indices,
item_size=self.indexer_page_stride_size,
dst_layout_dim=self.indexer_layout_dim,
num_layers=self.layer_num,
)
else:
raise ValueError(f"Unsupported layout: {self.layout}")
else: else:
transfer_kv_direct( if self.layout == "layer_first":
src_layers=device_pool.index_k_with_scale_buffer, transfer_kv_direct(
dst_layers=self.index_k_with_scale_buffer, src_layers=device_pool.index_k_with_scale_buffer,
src_indices=device_page_indices, dst_layers=self.index_k_with_scale_buffer,
dst_indices=host_page_indices, src_indices=device_page_indices,
page_size=1, dst_indices=host_page_indices,
) page_size=1,
)
elif self.layout == "page_first_direct":
transfer_kv_all_layer_direct_lf_pf(
src_ptrs=device_pool.index_k_with_scale_buffer,
dst_ptrs=[self.index_k_with_scale_buffer],
src_indices=device_page_indices,
dst_indices=host_page_indices,
page_size=1,
)
else:
raise ValueError(f"Unsupported layout: {self.layout}")
def load_to_device_per_layer( def load_to_device_per_layer(
self, self,