mod local_transfers;
#[allow(unused_macros)]
macro_rules! skip_if_stubs {
() => {
if kvbm_kernels::is_using_stubs() {
eprintln!(
"Skipping test '{}': stub kernels in use (no real CUDA)",
module_path!()
);
return;
}
};
}
#[allow(unused_macros)]
macro_rules! skip_if_stubs_and_device {
($($kind:expr),+ $(,)?) => {
if kvbm_kernels::is_using_stubs() {
let needs_cuda = false $(|| matches!($kind, StorageKind::Device(_)))+;
if needs_cuda {
eprintln!(
"Skipping test '{}': stub kernels in use and test requires Device storage",
module_path!()
);
return Ok(());
}
}
};
}
#[allow(unused_imports)]
pub(crate) use skip_if_stubs;
#[allow(unused_imports)]
pub(crate) use skip_if_stubs_and_device;
use super::{
BlockChecksum, FillPattern, NixlAgent, PhysicalLayout, StorageKind, TransferCapabilities,
compute_block_checksums, compute_layer_checksums, fill_blocks, fill_layers,
};
use crate::{
BlockId,
layout::{
BlockDimension, LayoutConfig,
builder::{HasConfig, NoLayout, NoMemory, PhysicalLayoutBuilder},
},
};
use anyhow::Result;
use std::collections::HashMap;
use std::ops::Range;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LayoutKind {
FC,
LW,
}
#[derive(Debug, Clone, Copy)]
pub struct LayoutSpec {
pub kind: LayoutKind,
pub storage: StorageKind,
}
impl LayoutSpec {
pub fn new(kind: LayoutKind, storage: StorageKind) -> Self {
Self { kind, storage }
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransferMode {
FullBlocks,
FirstLayerOnly,
SecondLayerOnly,
}
impl TransferMode {
pub fn layer_range(&self) -> Option<Range<usize>> {
match self {
TransferMode::FullBlocks => None,
TransferMode::FirstLayerOnly => Some(0..1),
TransferMode::SecondLayerOnly => Some(1..2),
}
}
pub fn suffix(&self) -> &'static str {
match self {
TransferMode::FullBlocks => "full",
TransferMode::FirstLayerOnly => "layer0",
TransferMode::SecondLayerOnly => "layer1",
}
}
}
pub fn standard_config(num_blocks: usize) -> LayoutConfig {
LayoutConfig::builder()
.num_blocks(num_blocks)
.num_layers(2)
.outer_dim(2)
.page_size(16)
.inner_dim(128)
.dtype_width_bytes(2)
.build()
.unwrap()
}
pub fn builder(num_blocks: usize) -> PhysicalLayoutBuilder<HasConfig, NoLayout, NoMemory> {
let agent = create_test_agent("test_agent");
let config = standard_config(num_blocks);
PhysicalLayout::builder(agent).with_config(config)
}
pub fn create_test_agent(name: &str) -> NixlAgent {
NixlAgent::new(name).expect("Failed to create agent")
}
#[expect(dead_code)]
pub fn create_test_agent_with_backends(name: &str, backends: &[&str]) -> Result<NixlAgent> {
NixlAgent::with_backends(name, backends)
}
pub fn create_fc_layout(
agent: NixlAgent,
storage_kind: StorageKind,
num_blocks: usize,
) -> PhysicalLayout {
let config = standard_config(num_blocks);
let builder = PhysicalLayout::builder(agent)
.with_config(config)
.fully_contiguous();
match storage_kind {
StorageKind::System => builder.allocate_system().build().unwrap(),
StorageKind::Pinned => builder.allocate_pinned(None).build().unwrap(),
StorageKind::Device(device_id) => builder.allocate_device(device_id).build().unwrap(),
StorageKind::Disk(_) => builder.allocate_disk(None).build().unwrap(),
}
}
pub fn create_lw_layout(
agent: NixlAgent,
storage_kind: StorageKind,
num_blocks: usize,
) -> PhysicalLayout {
let config = standard_config(num_blocks);
let builder = PhysicalLayout::builder(agent)
.with_config(config)
.layer_separate(BlockDimension::BlockIsFirstDim);
match storage_kind {
StorageKind::System => builder.allocate_system().build().unwrap(),
StorageKind::Pinned => builder.allocate_pinned(None).build().unwrap(),
StorageKind::Device(device_id) => builder.allocate_device(device_id).build().unwrap(),
StorageKind::Disk(_) => builder.allocate_disk(None).build().unwrap(),
}
}
pub fn create_layout(agent: NixlAgent, spec: LayoutSpec, num_blocks: usize) -> PhysicalLayout {
match spec.kind {
LayoutKind::FC => create_fc_layout(agent, spec.storage, num_blocks),
LayoutKind::LW => create_lw_layout(agent, spec.storage, num_blocks),
}
}
pub fn create_transfer_context(
agent: NixlAgent,
capabilities: Option<TransferCapabilities>,
) -> Result<crate::manager::TransferManager> {
crate::manager::TransferManager::builder()
.capabilities(capabilities.unwrap_or_default())
.nixl_agent(agent)
.cuda_device_id(0)
.build()
}
pub fn fill_and_checksum(
layout: &PhysicalLayout,
block_ids: &[BlockId],
pattern: FillPattern,
) -> Result<HashMap<BlockId, BlockChecksum>> {
fill_blocks(layout, block_ids, pattern)?;
compute_block_checksums(layout, block_ids)
}
pub fn fill_and_checksum_with_mode(
layout: &PhysicalLayout,
block_ids: &[BlockId],
pattern: FillPattern,
mode: TransferMode,
) -> Result<HashMap<BlockId, BlockChecksum>> {
match mode {
TransferMode::FullBlocks => {
fill_blocks(layout, block_ids, pattern)?;
compute_block_checksums(layout, block_ids)
}
TransferMode::FirstLayerOnly => {
fill_layers(layout, block_ids, 0..1, pattern)?;
compute_layer_checksums(layout, block_ids, 0..1)
}
TransferMode::SecondLayerOnly => {
fill_layers(layout, block_ids, 1..2, pattern)?;
compute_layer_checksums(layout, block_ids, 1..2)
}
}
}
pub fn verify_checksums_by_position(
src_checksums: &HashMap<BlockId, BlockChecksum>,
src_block_ids: &[BlockId],
dst_layout: &PhysicalLayout,
dst_block_ids: &[BlockId],
) -> Result<()> {
assert_eq!(
src_block_ids.len(),
dst_block_ids.len(),
"Source and destination block arrays must have same length"
);
let dst_checksums = compute_block_checksums(dst_layout, dst_block_ids)?;
for (src_id, dst_id) in src_block_ids.iter().zip(dst_block_ids.iter()) {
let src_checksum = src_checksums
.get(src_id)
.unwrap_or_else(|| panic!("Missing source checksum for block {}", src_id));
let dst_checksum = dst_checksums
.get(dst_id)
.unwrap_or_else(|| panic!("Missing destination checksum for block {}", dst_id));
assert_eq!(
src_checksum, dst_checksum,
"Checksum mismatch: src[{}] != dst[{}]: {} != {}",
src_id, dst_id, src_checksum, dst_checksum
);
}
Ok(())
}
pub fn verify_checksums_by_position_with_mode(
src_checksums: &HashMap<BlockId, BlockChecksum>,
src_block_ids: &[BlockId],
dst_layout: &PhysicalLayout,
dst_block_ids: &[BlockId],
mode: TransferMode,
) -> Result<()> {
assert_eq!(
src_block_ids.len(),
dst_block_ids.len(),
"Source and destination block arrays must have same length"
);
let dst_checksums = match mode {
TransferMode::FullBlocks => compute_block_checksums(dst_layout, dst_block_ids)?,
TransferMode::FirstLayerOnly => compute_layer_checksums(dst_layout, dst_block_ids, 0..1)?,
TransferMode::SecondLayerOnly => compute_layer_checksums(dst_layout, dst_block_ids, 1..2)?,
};
for (src_id, dst_id) in src_block_ids.iter().zip(dst_block_ids.iter()) {
let src_checksum = src_checksums
.get(src_id)
.unwrap_or_else(|| panic!("Missing source checksum for block {}", src_id));
let dst_checksum = dst_checksums
.get(dst_id)
.unwrap_or_else(|| panic!("Missing destination checksum for block {}", dst_id));
assert_eq!(
src_checksum, dst_checksum,
"Checksum mismatch (mode={:?}): src[{}] != dst[{}]: {} != {}",
mode, src_id, dst_id, src_checksum, dst_checksum
);
}
Ok(())
}
pub fn create_guard_blocks(
layout: &PhysicalLayout,
guard_block_ids: &[usize],
pattern: FillPattern,
) -> Result<HashMap<usize, BlockChecksum>> {
fill_blocks(layout, guard_block_ids, pattern)?;
compute_block_checksums(layout, guard_block_ids)
}
pub fn verify_guard_blocks_unchanged(
layout: &PhysicalLayout,
guard_block_ids: &[usize],
expected_checksums: &HashMap<usize, BlockChecksum>,
) -> Result<()> {
let current_checksums = compute_block_checksums(layout, guard_block_ids)?;
for &block_id in guard_block_ids {
let expected = expected_checksums
.get(&block_id)
.unwrap_or_else(|| panic!("Missing expected checksum for guard block {}", block_id));
let current = current_checksums
.get(&block_id)
.unwrap_or_else(|| panic!("Missing current checksum for guard block {}", block_id));
if expected != current {
return Err(anyhow::anyhow!(
"Guard block {} was modified during transfer! Expected: {}, Got: {}",
block_id,
expected,
current
));
}
}
Ok(())
}