diff --git a/.github/lsan-suppressions.txt b/.github/lsan-suppressions.txt new file mode 100644 index 0000000..e513f92 --- /dev/null +++ b/.github/lsan-suppressions.txt @@ -0,0 +1,6 @@ + +# Intel GPU OpenCL driver initialization allocations. +leak:libigdrcl.so + +# Intel OpenCL adapter queue/event/context leaks reported at process exit. +leak:libur_adapter_opencl.so diff --git a/.github/tsan-suppressions.txt b/.github/tsan-suppressions.txt new file mode 100644 index 0000000..e5c071a --- /dev/null +++ b/.github/tsan-suppressions.txt @@ -0,0 +1,22 @@ +# Prebuilt Intel CPU OpenCL runtime: worker mutex publication and JIT memset +# versus allocation. Reproduced with a pure C++ queue.memset(...).wait(). +# The exact libc++ mutex symbol is bundled in libintelocl; TSan cannot restrict +# a symbol match to that module, so this also excludes other libc++ mutex::lock calls. +race:^std::__1::mutex::lock()$ +race:^Intel::OpenCL::Utils::clAllocateFromHeap( + +# Prebuilt SYCL runtime: host-task completion versus event/queue destruction. +# Reproduced in pure C++ even with queue.wait() before destruction. Internal +# atomic shared_ptr synchronization is not visible to TSan in these binaries. +race:^sycl::_V1::detail::Scheduler::NotifyHostTaskCompletion( +race:^sycl::_V1::detail::event_impl::cleanupDependencyEvents( +race:^sycl::_V1::detail::Command::~Command( + +# Native-only reproduction: copy a host-task event, wait, then release the queue +# and event copies. The native event destructor frees a shared_ptr control block +# whose preceding reference-count decrement was in instrumented application code. +race:^sycl::_V1::detail::event_impl::~event_impl( + +# Keep these symbol-scoped: never suppress all of libsycl, the OpenCL adapter, +# free, memset, pthread calls, or the Rust/C++ shim. Revalidate on toolkit updates +# and remove entries when TSan-instrumented native runtimes are available. diff --git a/.github/workflows/basic.yml b/.github/workflows/basic.yml index a126f0e..713fba3 100644 --- a/.github/workflows/basic.yml +++ b/.github/workflows/basic.yml @@ -86,6 +86,69 @@ jobs: name: documentation path: target/doc + hardening: + if: github.event_name != 'schedule' + runs-on: ["oneapi-rs", "Linux"] + + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + + - name: Setup oneAPI + uses: ./.github/actions/setup-oneapi + + - name: Install nightly toolchain + run: rustup toolchain install nightly --profile minimal + + - name: Build and test with hardening flags + env: + RUSTFLAGS: -Zemit-stack-sizes -Zcf-protection=full + CARGO_TARGET_DIR: target/hardened + run: | + cargo +nightly test --workspace --verbose + cargo +nightly build --workspace --release --verbose + + sanitizers: + name: Sanitizer (${{ matrix.sanitizer }}) + if: github.event_name != 'schedule' + runs-on: ["oneapi-rs", "Linux"] + strategy: + fail-fast: false + matrix: + sanitizer: [address, leak, thread] + env: + CARGO_TARGET_DIR: target/sanitizer-${{ matrix.sanitizer }} + + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + + - name: Setup oneAPI + uses: ./.github/actions/setup-oneapi + + - name: Install nightly toolchain + run: rustup toolchain install nightly --profile minimal + + - name: Install standard library sources for ThreadSanitizer + if: matrix.sanitizer == 'thread' + run: rustup component add rust-src --toolchain nightly + + - name: Run tests with sanitizers + env: + # Test only CPU to avoid issues with GPU drivers and sanitizers + ONEAPI_DEVICE_SELECTOR: opencl:cpu + # Leave leak detection to the dedicated leak sanitizer job. + ASAN_OPTIONS: detect_leaks=0 + LSAN_OPTIONS: suppressions=${{ github.workspace }}/.github/lsan-suppressions.txt + TSAN_OPTIONS: suppressions=${{ github.workspace }}/.github/tsan-suppressions.txt + RUSTFLAGS: -Zsanitizer=${{ matrix.sanitizer }} + run: | + if [ "${{ matrix.sanitizer }}" = thread ]; then + # Instrument the C++ host shim, not SYCL device code. + export CXX=icpx + export CXXFLAGS="-Xarch_host -fsanitize=thread" + fi + cargo +nightly test ${{ matrix.sanitizer == 'thread' && '-Zbuild-std' || '' }} \ + --workspace --lib --tests --target x86_64-unknown-linux-gnu --verbose + trivy: name: Trivy scan runs-on: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 446be5c..d53b4d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,8 @@ can be found at: - Initial Rust API for discovering SYCL platforms and devices. - Queues, events, contexts, USM allocations, and host/device memory operations. +- Dropping a `UsmBox` waits for the copies and fills that use it, so its + memory is not freed while a device may still use it. - Runtime kernel bundle compilation and one-, two-, and three-dimensional kernel launch support. - Derive support for typed kernel argument lists. diff --git a/sycl/sycl-rs/src/lib.rs b/sycl/sycl-rs/src/lib.rs index 0073685..bffcd59 100644 --- a/sycl/sycl-rs/src/lib.rs +++ b/sycl/sycl-rs/src/lib.rs @@ -90,9 +90,10 @@ //! ``` //! //! # Safety model -//! - USM allocations are represented by a zero-cost [`UsmBox`](crate::usmbox::UsmBox) type managed -//! through RAII. +//! - USM allocations are represented by a [`UsmBox`](crate::usmbox::UsmBox) type managed through +//! RAII. //! - Note: `UsmBox` arrays do not rely on accessors, unlike SYCL buffers. +//! - Dropping a `UsmBox` waits for the copies and fills that use it to complete. //! - `UsmBox`es are zero-initialized by default. //! - `UsmBox`es can only store types that implement [`bytemuck::Pod`]. //! - Kernel launch is inherently unsafe. In particular, the caller must ensure that every argument diff --git a/sycl/sycl-rs/src/queue.rs b/sycl/sycl-rs/src/queue.rs index d0814e4..2ff6f07 100644 --- a/sycl/sycl-rs/src/queue.rs +++ b/sycl/sycl-rs/src/queue.rs @@ -140,7 +140,10 @@ impl Queue { ptr: (*e).clone().0, }) .collect::>(); - unsafe { ffi::memset(&mut self.0, ptr, value, num_bytes, dep_events) }.map(Into::into) + let event: Event = + unsafe { ffi::memset(&mut self.0, ptr, value, num_bytes, dep_events) }?.into(); + array.add_pending(&event); + Ok(event) } /// Submits a barrier to the queue. @@ -228,7 +231,7 @@ impl Queue { .collect::>(); let num_bytes = src.get_len() * size_of::(); - unsafe { + let event: Event = unsafe { ffi::memcpy( &mut self.0, dst.get_byte_ptr(), @@ -236,8 +239,11 @@ impl Queue { num_bytes, dep_events, ) - } - .map(Into::into) + }? + .into(); + src.add_pending(&event); + dst.add_pending(&event); + Ok(event) } } diff --git a/sycl/sycl-rs/src/usmbox.rs b/sycl/sycl-rs/src/usmbox.rs index b9ba1ab..a90cac6 100644 --- a/sycl/sycl-rs/src/usmbox.rs +++ b/sycl/sycl-rs/src/usmbox.rs @@ -8,6 +8,7 @@ use std::{ alloc::{Layout, handle_alloc_error}, + cell::RefCell, ops::{Deref, DerefMut}, pin::Pin, ptr::NonNull, @@ -21,6 +22,7 @@ use pin_project::pin_project; use crate::{ Result, event::{Event, EventFuture}, + info::{self, InfoTarget}, kernel::KernelArgument, usm::{ DeviceAllocator, HostAccessible, HostAllocator, SharedAllocator, UsmAlloc, UsmAllocator, @@ -39,11 +41,16 @@ use crate::{ /// The `UsmBox` struct takes a generic parameter [`UsmAlloc`](`crate::usm::UsmAlloc`) for /// specifying an allocator which is used by the SYCL runtime when allocating temporary memory on /// the host. +/// +/// A `UsmBox` keeps track of the copies and fills that use it, and dropping it waits for them to +/// complete, so its memory is not freed while a device may still use it. pub struct UsmBox { data: NonNull, len: usize, layout: Layout, allocator: A, + /// Events of the enqueued commands that may still use the memory. + pending: RefCell>, } impl UsmBox { @@ -61,7 +68,25 @@ impl UsmBox { len, layout, allocator, + pending: RefCell::default(), + } + } + + /// Records an enqueued command that uses the memory, so that dropping the box waits for it. + pub(crate) fn add_pending(&self, event: &Event) { + let mut pending = self.pending.borrow_mut(); + // Forget the completed commands only once the list is full, and leave room for as many + // new ones as are still pending, so that recording a command takes amortized constant + // time even when many are in flight. + if pending.len() == pending.capacity() { + pending.retain(|command| { + command.get_info::() + != info::EventCommandStatus::Complete + }); + let still_pending = pending.len(); + pending.reserve(still_pending); } + pending.push(event.clone()); } pub(crate) fn get_byte_ptr(&self) -> *mut u8 { @@ -96,10 +121,15 @@ impl DerefMut for UsmBox { } } +/// Waits for the copies and fills that use the `UsmBox`, then frees its memory. If waiting fails, +/// the memory is leaked, since a command may still be using it. impl Drop for UsmBox { fn drop(&mut self) { - unsafe { - self.allocator.deallocate(self.data.cast(), self.layout); + let waited = self.pending.get_mut().iter_mut().try_for_each(Event::wait); + if waited.is_ok() { + unsafe { + self.allocator.deallocate(self.data.cast(), self.layout); + } } } } @@ -124,7 +154,7 @@ impl EnqueuedUsmBox { /// Performs a blocking wait for the [`UsmBox`] initialization to complete. Returns an error if /// a synchronous SYCL exception occurs. /// - /// Dropping an enqueued array does not wait for its completion. + /// Dropping an enqueued array waits for its initialization to complete, then frees it. pub fn wait(mut self) -> Result> { self.event.wait().map(|_| self.array) } @@ -185,3 +215,204 @@ unsafe impl KernelArgument for &mut UsmBox { unsafe { self.as_raw_arg_impl() } } } + +#[cfg(test)] +mod tests { + use super::*; + use allocator_api2::alloc::{AllocError, Allocator}; + use std::{ + sync::{ + Arc, Condvar, Mutex, + atomic::{AtomicBool, Ordering}, + mpsc, + }, + task::{Wake, Waker}, + thread, + time::Duration, + }; + use sycl_rs_sys::{event::ffi, types::SharedWaker}; + + #[derive(Default)] + struct Gate { + state: Mutex<(bool, bool)>, + changed: Condvar, + } + + impl Wake for Gate { + fn wake(self: Arc) { + let mut state = self.state.lock().unwrap(); + state.0 = true; + self.changed.notify_all(); + while !state.1 { + state = self.changed.wait(state).unwrap(); + } + } + } + + struct Release(Arc); + + impl Drop for Release { + fn drop(&mut self) { + self.0.state.lock().unwrap().1 = true; + self.0.changed.notify_all(); + } + } + + struct ObservedAllocator<'a> { + allocation: &'a HostUsmBox, + event: &'a Event, + deallocated: &'a AtomicBool, + premature: &'a AtomicBool, + } + + unsafe impl Allocator for ObservedAllocator<'_> { + fn allocate(&self, layout: Layout) -> std::result::Result, AllocError> { + assert_eq!(layout.size(), self.allocation.get_byte_size()); + Ok(NonNull::slice_from_raw_parts( + NonNull::new(self.allocation.get_byte_ptr()).unwrap(), + layout.size(), + )) + } + + unsafe fn deallocate(&self, _: NonNull, _: Layout) { + let complete = matches!( + ffi::get_command_execution_status(&self.event.0), + sycl_rs_sys::types::ffi::EventCommandStatus::Complete + ); + self.premature.store(!complete, Ordering::SeqCst); + self.deallocated.store(true, Ordering::SeqCst); + } + } + + unsafe impl UsmAlloc for ObservedAllocator<'_> {} + + #[derive(Clone, Copy)] + enum Completion { + DropEnqueued, + DropUnpolled, + DropPolled, + Wait, + Await, + } + + fn check_initialization_ownership(completion: Completion) -> Result<()> { + // Isolate fixtures from concurrent device discovery in the uninstrumented CPU runtime. + static FIXTURE: Mutex<()> = Mutex::new(()); + let fixture = FIXTURE.lock().unwrap(); + let mut queue = crate::queue::Queue::new(); + let dependency = queue.barrier()?; + let gate = Arc::new(Gate::default()); + let release = Release(gate.clone()); + let shared = Arc::new(SharedWaker::new()); + shared.waker.register(&Waker::from(gate.clone())); + unsafe { + ffi::register_callback(&mut queue.0, &dependency.0, Arc::into_raw(shared))?; + } + let (state, timeout) = gate + .changed + .wait_timeout_while(gate.state.lock().unwrap(), Duration::from_secs(10), |s| { + !s.0 + }) + .unwrap(); + assert!(!timeout.timed_out(), "initialization gate did not start"); + drop(state); + + let mut allocation = unsafe { queue.alloc_uninit_host::(16) }; + let event = unsafe { queue.memset(&mut allocation, 0) }?; + assert!(!matches!( + ffi::get_command_execution_status(&event.0), + sycl_rs_sys::types::ffi::EventCommandStatus::Complete + )); + let deallocated = AtomicBool::new(false); + let premature = AtomicBool::new(false); + // Keep real backing memory alive even if the ownership regression reappears. + let array = unsafe { + UsmBox::::new( + ObservedAllocator { + allocation: &allocation, + event: &event, + deallocated: &deallocated, + premature: &premature, + }, + 16, + ) + }; + let enqueued = EnqueuedUsmBox::new(array, event.clone()); + + thread::scope(|scope| -> Result<()> { + let (start, started) = mpsc::channel(); + scope.spawn(move || { + let _release = release; + started + .recv_timeout(Duration::from_secs(10)) + .expect("completion path did not start"); + thread::sleep(Duration::from_millis(100)); + }); + match completion { + Completion::DropEnqueued => { + start.send(()).unwrap(); + drop(enqueued); + } + Completion::DropUnpolled => { + let future = enqueued.into_future(); + start.send(()).unwrap(); + drop(future); + } + Completion::DropPolled => { + let mut future = Box::pin(enqueued.into_future()); + let mut context = Context::from_waker(Waker::noop()); + assert!(future.as_mut().poll(&mut context).is_pending()); + start.send(()).unwrap(); + drop(future); + } + Completion::Wait => { + start.send(()).unwrap(); + let array = enqueued.wait()?; + assert!(!deallocated.load(Ordering::SeqCst)); + drop(array); + } + Completion::Await => { + start.send(()).unwrap(); + let array = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + .block_on(enqueued.into_future())?; + assert!(!deallocated.load(Ordering::SeqCst)); + drop(array); + } + } + Ok(()) + })?; + queue.wait()?; + drop(fixture); + assert!(deallocated.load(Ordering::SeqCst)); + assert!(!premature.load(Ordering::SeqCst)); + assert!(allocation.iter().all(|&value| value == 0)); + Ok(()) + } + + #[test] + fn dropping_enqueued_allocation_waits_for_initialization() -> Result<()> { + check_initialization_ownership(Completion::DropEnqueued) + } + + #[test] + fn dropping_unpolled_allocation_future_waits_for_initialization() -> Result<()> { + check_initialization_ownership(Completion::DropUnpolled) + } + + #[test] + fn dropping_polled_allocation_future_waits_for_initialization() -> Result<()> { + check_initialization_ownership(Completion::DropPolled) + } + + #[test] + fn waiting_returns_initialized_allocation() -> Result<()> { + check_initialization_ownership(Completion::Wait) + } + + #[test] + fn awaiting_returns_initialized_allocation() -> Result<()> { + check_initialization_ownership(Completion::Await) + } +}