#[cfg(feature = "cuda")]
use std::sync::Arc;
#[cfg(feature = "cuda")]
use crate::expert_slots::{SlotDevice, SlotGeometry};
#[cfg(feature = "cuda")]
use crate::residency::CopyRoute;
#[cfg(feature = "cuda")]
use cudarc::driver::{CudaDevice, CudaSlice, DeviceSlice};
pub fn split_pair<T>(slots: &mut [T], dst: usize, src: usize) -> Option<(&T, &mut T)> {
if dst == src || dst.max(src) >= slots.len() {
return None;
}
let (low, high) = slots.split_at_mut(dst.max(src));
if dst < src {
let target = &mut low[dst];
Some((&high[0], target))
} else {
Some((&low[src], &mut high[0]))
}
}
#[cfg(feature = "cuda")]
pub struct CudaExpertPool {
dev: Arc<CudaDevice>,
banks: Vec<Vec<CudaSlice<u8>>>,
row_bytes: Vec<usize>,
}
#[cfg(feature = "cuda")]
impl CudaExpertPool {
pub fn new(dev: Arc<CudaDevice>, geometry: &SlotGeometry) -> Result<Self, String> {
let mut banks = Vec::with_capacity(geometry.banks());
for &row_bytes in &geometry.row_bytes {
let mut slots = Vec::with_capacity(geometry.slots);
for slot in 0..geometry.slots {
slots.push(dev.alloc_zeros::<u8>(row_bytes).map_err(|e| {
format!(
"allocating slot {slot} of {} x {row_bytes} bytes: {e:?}",
geometry.slots
)
})?);
}
banks.push(slots);
}
Ok(CudaExpertPool {
dev,
banks,
row_bytes: geometry.row_bytes.clone(),
})
}
pub fn slot(&self, bank: usize, slot: u32) -> Option<&CudaSlice<u8>> {
self.banks.get(bank)?.get(slot as usize)
}
pub fn bytes(&self) -> u64 {
self.row_bytes
.iter()
.zip(self.banks.iter())
.map(|(w, slots)| *w as u64 * slots.len() as u64)
.sum()
}
}
#[cfg(feature = "cuda")]
impl SlotDevice for CudaExpertPool {
fn begin_plan(&mut self, route: CopyRoute) -> Result<(), String> {
let _ = route;
Ok(())
}
fn write_slot(&mut self, bank: usize, dst_slot: u32, src: &[u8]) -> Result<(), String> {
let dev = Arc::clone(&self.dev);
let dst = self
.banks
.get_mut(bank)
.and_then(|b| b.get_mut(dst_slot as usize))
.ok_or_else(|| format!("no slot {dst_slot} in bank {bank}"))?;
if src.len() != dst.len() {
return Err(format!(
"bank {bank} slot {dst_slot} is {} bytes, but the row offered is {}",
dst.len(),
src.len()
));
}
dev.htod_sync_copy_into(src, dst)
.map_err(|e| format!("host-to-device copy into bank {bank} slot {dst_slot}: {e:?}"))
}
fn copy_slot(&mut self, bank: usize, dst_slot: u32, src_slot: u32) -> Result<(), String> {
if dst_slot == src_slot {
return Ok(());
}
let dev = Arc::clone(&self.dev);
let slots = self
.banks
.get_mut(bank)
.ok_or_else(|| format!("no bank {bank}"))?;
let len = slots.len();
let (source, target) =
split_pair(slots, dst_slot as usize, src_slot as usize).ok_or_else(|| {
format!("bank {bank} holds {len} slots, not {dst_slot} <- {src_slot}")
})?;
dev.dtod_copy(source, target).map_err(|e| {
format!("device-to-device copy {src_slot} -> {dst_slot} in bank {bank}: {e:?}")
})
}
fn flush(&mut self) -> Result<(), String> {
Ok(())
}
}
#[cfg(test)]
mod split_tests {
use super::split_pair;
#[test]
fn the_split_hands_back_the_source_and_destination_the_caller_asked_for() {
for (dst, src) in [(0usize, 3usize), (3, 0), (1, 2), (2, 1), (0, 1), (3, 2)] {
let mut slots: Vec<u32> = (0..4).collect();
let (source, target) = split_pair(&mut slots, dst, src).expect("distinct, in range");
assert_eq!(*source, src as u32, "source for {dst} <- {src}");
assert_eq!(*target, dst as u32, "destination for {dst} <- {src}");
*target = *source;
assert_eq!(slots[dst], src as u32, "the copy landed in {dst}");
assert_eq!(slots[src], src as u32, "and left the source alone");
}
}
#[test]
fn a_self_copy_or_an_out_of_range_slot_is_not_a_pair() {
let mut slots: Vec<u32> = (0..4).collect();
assert!(split_pair(&mut slots, 2, 2).is_none());
assert!(split_pair(&mut slots, 4, 0).is_none());
assert!(split_pair(&mut slots, 0, 4).is_none());
assert!(split_pair::<u32>(&mut [], 0, 1).is_none());
}
}
#[cfg(all(test, feature = "cuda"))]
mod tests {
use super::*;
use crate::expert_cache::{CopyPlan, ExpertId, GatherPlan};
use crate::expert_slots::{ExpertRows, ExpertSlots};
struct NamedRows {
rows: Vec<Vec<u8>>,
}
impl NamedRows {
fn new(experts: usize, width: usize) -> Self {
let rows = (0..experts)
.map(|e| (0..width).map(|i| (e as u8) << 4 | i as u8).collect())
.collect();
NamedRows { rows }
}
}
impl ExpertRows for NamedRows {
fn row(&self, bank: usize, _layer: u32, row: u32) -> Option<&[u8]> {
if bank > 0 {
return None;
}
self.rows.get(row as usize).map(|r| r.as_slice())
}
}
#[test]
#[ignore = "requires real CUDA hardware -- NOT yet run on a GPU; run with --ignored on a CUDA-capable machine"]
fn a_cuda_pool_lands_each_expert_in_the_slot_the_plan_named() {
const EXPERTS: usize = 4;
const WIDTH: usize = 64;
let dev = CudaDevice::new(0).expect("a CUDA device");
let geometry = SlotGeometry {
num_layers: 1,
slots: EXPERTS,
row_bytes: vec![WIDTH],
};
let mut pool = CudaExpertPool::new(dev.clone(), &geometry).unwrap();
assert_eq!(pool.bytes(), (EXPERTS * WIDTH) as u64);
let mut slots = ExpertSlots::new(geometry).unwrap();
let rows = NamedRows::new(EXPERTS, WIDTH);
let plan = CopyPlan {
dst_slots: vec![0, 2],
src_rows: vec![3, 1],
};
let applied = slots.apply_copy_plan(0, &plan, &rows, &mut pool).unwrap();
assert_eq!(applied.rows, 2);
assert_eq!(applied.bytes as usize, 2 * WIDTH);
for (&slot, &row) in plan.dst_slots.iter().zip(plan.src_rows.iter()) {
let got = dev.dtoh_sync_copy(pool.slot(0, slot).unwrap()).unwrap();
assert_eq!(got, rows.rows[row as usize], "slot {slot} holds row {row}");
}
let gather = GatherPlan {
dst_slots: vec![1],
src_slots: vec![0],
};
slots.apply_gather_plan(&gather, &mut pool).unwrap();
assert_eq!(
dev.dtoh_sync_copy(pool.slot(0, 1).unwrap()).unwrap(),
rows.rows[3],
"the gather must carry slot 0's contents, not its neighbour's"
);
assert_eq!(
slots.occupant(1),
Some(ExpertId {
layer: 0,
expert: 3
})
);
let stats = slots.stats();
assert_eq!(stats.host_bytes as usize, 2 * WIDTH);
assert_eq!(
stats.device_bytes as usize, WIDTH,
"a device-to-device copy must not be billed to the link"
);
let warm = slots
.apply_copy_plan(0, &CopyPlan::default(), &rows, &mut pool)
.unwrap();
assert!(warm.warm && warm.bytes == 0);
assert_eq!(slots.stats().host_bytes, stats.host_bytes);
}
}