use core::marker::PhantomData;
use std::sync::Arc;
use crate::driver::{result, sys};
use super::{
CudaContext, CudaEvent, CudaStream, DevicePtr, DevicePtrMut, DeviceRepr, DeviceSlice,
DriverError, HostSlice, LaunchArgs, PushKernelArg, ValidAsZeroBits,
};
#[derive(Debug)]
pub struct UnifiedSlice<T> {
pub(crate) cu_device_ptr: sys::CUdeviceptr,
pub(crate) len: usize,
pub(crate) stream: Arc<CudaStream>,
pub(crate) event: CudaEvent,
pub(crate) attach_mode: sys::CUmemAttach_flags,
pub(crate) concurrent_managed_access: bool,
pub(crate) marker: PhantomData<*const T>,
}
unsafe impl<T> Send for UnifiedSlice<T> {}
unsafe impl<T> Sync for UnifiedSlice<T> {}
impl<T> Drop for UnifiedSlice<T> {
fn drop(&mut self) {
self.stream.ctx.record_err(self.event.synchronize());
self.stream
.ctx
.record_err(unsafe { result::memory_free(self.cu_device_ptr) });
}
}
impl CudaContext {
pub unsafe fn alloc_unified<T: DeviceRepr>(
self: &Arc<Self>,
len: usize,
attach_global: bool,
) -> Result<UnifiedSlice<T>, DriverError> {
if self.attribute(sys::CUdevice_attribute::CU_DEVICE_ATTRIBUTE_MANAGED_MEMORY)? == 0 {
return Err(DriverError(sys::cudaError_enum::CUDA_ERROR_NOT_PERMITTED));
}
let attach_mode = if attach_global {
sys::CUmemAttach_flags::CU_MEM_ATTACH_GLOBAL
} else {
sys::CUmemAttach_flags::CU_MEM_ATTACH_HOST
};
let cu_device_ptr = result::malloc_managed(len * std::mem::size_of::<T>(), attach_mode)?;
let concurrent_managed_access = self
.attribute(sys::CUdevice_attribute::CU_DEVICE_ATTRIBUTE_CONCURRENT_MANAGED_ACCESS)?
!= 0;
let stream = self.default_stream();
let event = self.new_event(Some(sys::CUevent_flags::CU_EVENT_BLOCKING_SYNC))?;
Ok(UnifiedSlice {
cu_device_ptr,
len,
stream,
event,
attach_mode,
concurrent_managed_access,
marker: PhantomData,
})
}
}
impl<T> UnifiedSlice<T> {
pub fn len(&self) -> usize {
self.len
}
pub fn is_empty(&self) -> bool {
self.len == 0
}
pub fn attach_mode(&self) -> sys::CUmemAttach_flags {
self.attach_mode
}
pub fn num_bytes(&self) -> usize {
self.len * std::mem::size_of::<T>()
}
pub fn attach(
&mut self,
stream: &Arc<CudaStream>,
flags: sys::CUmemAttach_flags,
) -> Result<(), DriverError> {
self.event.synchronize()?;
self.stream = stream.clone();
self.attach_mode = flags;
unsafe {
result::stream::attach_mem_async(
self.stream.cu_stream,
self.cu_device_ptr,
self.num_bytes(),
self.attach_mode,
)
}
}
#[cfg(not(any(
feature = "cuda-11040",
feature = "cuda-11050",
feature = "cuda-11060",
feature = "cuda-11070",
feature = "cuda-11080",
feature = "cuda-12000",
feature = "cuda-12010"
)))]
pub fn prefetch(&self) -> Result<(), DriverError> {
let location = match self.attach_mode {
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_GLOBAL
| sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_SINGLE => {
if !self.concurrent_managed_access {
return Err(DriverError(sys::cudaError_enum::CUDA_ERROR_NOT_PERMITTED));
}
sys::CUmemLocation {
type_: sys::CUmemLocationType::CU_MEM_LOCATION_TYPE_DEVICE,
id: self.stream.ctx.ordinal as i32,
}
}
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_HOST => {
sys::CUmemLocation {
type_: sys::CUmemLocationType::CU_MEM_LOCATION_TYPE_HOST_NUMA_CURRENT,
id: 0, }
}
};
unsafe {
result::mem_prefetch_async(
self.cu_device_ptr,
self.len * std::mem::size_of::<T>(),
location,
self.stream.cu_stream,
)
}
}
pub fn check_host_access(&self) -> Result<(), DriverError> {
match self.attach_mode {
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_GLOBAL => {
}
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_HOST => {
}
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_SINGLE => {
self.stream.synchronize()?;
}
};
Ok(())
}
pub fn check_device_access(&self, stream: &CudaStream) -> Result<(), DriverError> {
match self.attach_mode {
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_GLOBAL => {
}
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_HOST => {
let concurrent_managed_access = if self.stream.context() != stream.context() {
stream.context().attribute(
sys::CUdevice_attribute::CU_DEVICE_ATTRIBUTE_CONCURRENT_MANAGED_ACCESS,
)? != 0
} else {
self.concurrent_managed_access
};
if !concurrent_managed_access {
return Err(DriverError(sys::cudaError_enum::CUDA_ERROR_NOT_PERMITTED));
}
}
sys::CUmemAttach_flags_enum::CU_MEM_ATTACH_SINGLE => {
if self.stream.as_ref() != stream {
return Err(DriverError(sys::cudaError_enum::CUDA_ERROR_NOT_PERMITTED));
}
}
};
Ok(())
}
}
impl<T> DeviceSlice<T> for UnifiedSlice<T> {
fn len(&self) -> usize {
self.len
}
fn stream(&self) -> &Arc<CudaStream> {
&self.stream
}
}
impl<T> DevicePtr<T> for UnifiedSlice<T> {
fn device_ptr<'a>(
&'a self,
stream: &'a CudaStream,
) -> (sys::CUdeviceptr, super::SyncOnDrop<'a>) {
stream.ctx.record_err(self.check_device_access(stream));
stream.ctx.record_err(stream.wait(&self.event));
(
self.cu_device_ptr,
super::SyncOnDrop::Record(Some((&self.event, stream))),
)
}
}
impl<T> DevicePtrMut<T> for UnifiedSlice<T> {
fn device_ptr_mut<'a>(
&'a mut self,
stream: &'a CudaStream,
) -> (sys::CUdeviceptr, super::SyncOnDrop<'a>) {
stream.ctx.record_err(self.check_device_access(stream));
stream.ctx.record_err(stream.wait(&self.event));
(
self.cu_device_ptr,
super::SyncOnDrop::Record(Some((&self.event, stream))),
)
}
}
impl<T: ValidAsZeroBits> UnifiedSlice<T> {
pub fn as_slice(&self) -> Result<&[T], DriverError> {
self.check_host_access()?;
self.event.synchronize()?;
Ok(unsafe { std::slice::from_raw_parts(self.cu_device_ptr as *const T, self.len) })
}
pub fn as_mut_slice(&mut self) -> Result<&mut [T], DriverError> {
self.check_host_access()?;
self.event.synchronize()?;
Ok(unsafe { std::slice::from_raw_parts_mut(self.cu_device_ptr as *mut T, self.len) })
}
}
impl<T> HostSlice<T> for UnifiedSlice<T> {
fn len(&self) -> usize {
self.len
}
unsafe fn stream_synced_slice<'a>(
&'a self,
stream: &'a CudaStream,
) -> (&'a [T], super::SyncOnDrop<'a>) {
stream.ctx.record_err(self.check_device_access(stream));
stream.ctx.record_err(stream.wait(&self.event));
(
std::slice::from_raw_parts(self.cu_device_ptr as *const T, self.len),
super::SyncOnDrop::Record(Some((&self.event, stream))),
)
}
unsafe fn stream_synced_mut_slice<'a>(
&'a mut self,
stream: &'a CudaStream,
) -> (&'a mut [T], super::SyncOnDrop<'a>) {
stream.ctx.record_err(self.check_device_access(stream));
stream.ctx.record_err(stream.wait(&self.event));
(
std::slice::from_raw_parts_mut(self.cu_device_ptr as *mut T, self.len),
super::SyncOnDrop::Record(Some((&self.event, stream))),
)
}
}
unsafe impl<'a, 'b: 'a, T> PushKernelArg<&'b UnifiedSlice<T>> for LaunchArgs<'a> {
#[inline(always)]
fn arg(&mut self, arg: &'b UnifiedSlice<T>) -> &mut Self {
self.stream
.ctx
.record_err(arg.check_device_access(self.stream));
self.waits.push(&arg.event);
self.records.push(&arg.event);
self.args
.push((&arg.cu_device_ptr) as *const sys::CUdeviceptr as _);
self
}
}
unsafe impl<'a, 'b: 'a, T> PushKernelArg<&'b mut UnifiedSlice<T>> for LaunchArgs<'a> {
#[inline(always)]
fn arg(&mut self, arg: &'b mut UnifiedSlice<T>) -> &mut Self {
self.stream
.ctx
.record_err(arg.check_device_access(self.stream));
self.waits.push(&arg.event);
self.records.push(&arg.event);
self.args
.push((&arg.cu_device_ptr) as *const sys::CUdeviceptr as _);
self
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::needless_range_loop)]
use crate::driver::{LaunchConfig, PushKernelArg};
use super::*;
#[test]
fn test_unified_memory_global() -> Result<(), DriverError> {
let ctx = CudaContext::new(0)?;
let mut a = unsafe { ctx.alloc_unified::<f32>(100, true) }?;
{
let buf = a.as_mut_slice()?;
for i in 0..100 {
buf[i] = i as f32;
}
}
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], i as f32);
}
}
let ptx = crate::nvrtc::compile_ptx(
"
extern \"C\" __global__ void kernel(float *buf) {
if (threadIdx.x < 100) {
assert(buf[threadIdx.x] == static_cast<float>(threadIdx.x));
}
}",
)
.unwrap();
let module = ctx.load_module(ptx)?;
let f = module.load_function("kernel")?;
let stream1 = ctx.default_stream();
unsafe {
stream1
.launch_builder(&f)
.arg(&mut a)
.launch(LaunchConfig::for_num_elems(100))
}?;
stream1.synchronize()?;
let stream2 = ctx.new_stream()?;
unsafe {
stream2
.launch_builder(&f)
.arg(&mut a)
.launch(LaunchConfig::for_num_elems(100))
}?;
stream2.synchronize()?;
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], i as f32);
}
}
let vs = stream1.memcpy_dtov(&a)?;
for i in 0..100 {
assert_eq!(vs[i], i as f32);
}
let b = stream1.memcpy_stod(&a)?;
let vs = stream1.memcpy_dtov(&b)?;
for i in 0..100 {
assert_eq!(vs[i], i as f32);
}
stream1.memset_zeros(&mut a)?;
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], 0.0);
}
}
Ok(())
}
#[test]
fn test_unified_memory_host() -> Result<(), DriverError> {
let ctx = CudaContext::new(0)?;
let mut a = unsafe { ctx.alloc_unified::<f32>(100, false) }?;
{
let buf = a.as_mut_slice()?;
for i in 0..100 {
buf[i] = i as f32;
}
}
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], i as f32);
}
}
let ptx = crate::nvrtc::compile_ptx(
"
extern \"C\" __global__ void kernel(float *buf) {
if (threadIdx.x < 100) {
assert(buf[threadIdx.x] == static_cast<float>(threadIdx.x));
}
}",
)
.unwrap();
let module = ctx.load_module(ptx)?;
let f = module.load_function("kernel")?;
let stream1 = ctx.default_stream();
unsafe {
stream1
.launch_builder(&f)
.arg(&mut a)
.launch(LaunchConfig::for_num_elems(100))
}?;
stream1.synchronize()?;
let stream2 = ctx.new_stream()?;
unsafe {
stream2
.launch_builder(&f)
.arg(&mut a)
.launch(LaunchConfig::for_num_elems(100))
}?;
stream2.synchronize()?;
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], i as f32);
}
}
let vs = stream1.memcpy_dtov(&a)?;
for i in 0..100 {
assert_eq!(vs[i], i as f32);
}
let b = stream1.memcpy_stod(&a)?;
let vs = stream1.memcpy_dtov(&b)?;
for i in 0..100 {
assert_eq!(vs[i], i as f32);
}
stream1.memset_zeros(&mut a)?;
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], 0.0);
}
}
Ok(())
}
#[test]
fn test_unified_memory_single_stream() -> Result<(), DriverError> {
let ctx = CudaContext::new(0)?;
let mut a = unsafe { ctx.alloc_unified::<f32>(100, true) }?;
{
let buf = a.as_mut_slice()?;
for i in 0..100 {
buf[i] = i as f32;
}
}
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], i as f32);
}
}
let ptx = crate::nvrtc::compile_ptx(
"
extern \"C\" __global__ void kernel(float *buf) {
if (threadIdx.x < 100) {
assert(buf[threadIdx.x] == static_cast<float>(threadIdx.x));
}
}",
)
.unwrap();
let module = ctx.load_module(ptx)?;
let f = module.load_function("kernel")?;
let stream2 = ctx.new_stream()?;
a.attach(&stream2, sys::CUmemAttach_flags::CU_MEM_ATTACH_SINGLE)?;
unsafe {
stream2
.launch_builder(&f)
.arg(&mut a)
.launch(LaunchConfig::for_num_elems(100))
}?;
stream2.synchronize()?;
let stream1 = ctx.default_stream();
unsafe {
stream1
.launch_builder(&f)
.arg(&mut a)
.launch(LaunchConfig::for_num_elems(100))
}
.expect_err("Other stream access should've failed");
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], i as f32);
}
}
let vs = stream2.memcpy_dtov(&a)?;
for i in 0..100 {
assert_eq!(vs[i], i as f32);
}
let b = stream2.memcpy_stod(&a)?;
let vs = stream2.memcpy_dtov(&b)?;
for i in 0..100 {
assert_eq!(vs[i], i as f32);
}
stream2.memset_zeros(&mut a)?;
{
let buf = a.as_slice()?;
for i in 0..100 {
assert_eq!(buf[i], 0.0);
}
}
Ok(())
}
}