From 5d226f5d87f5c91133e88bc20ba20de5982aeef4 Mon Sep 17 00:00:00 2001 From: Rafal Rudnicki Date: Fri, 16 Oct 2026 13:09:39 +0000 Subject: [PATCH 1/2] Add sanitizers workflow --- .github/lsan-suppressions.txt | 6 ++++ .github/workflows/basic.yml | 57 +++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+) create mode 100644 .github/lsan-suppressions.txt 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/workflows/basic.yml b/.github/workflows/basic.yml index a126f0e..2ec476e 100644 --- a/.github/workflows/basic.yml +++ b/.github/workflows/basic.yml @@ -86,6 +86,63 @@ 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 + RUSTFLAGS: -Zsanitizer=${{ matrix.sanitizer }} + run: | + 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 From beae19dc9aeacdd3968947489e6e06241c547126 Mon Sep 17 00:00:00 2001 From: Rafal Rudnicki Date: Thu, 29 Oct 2026 18:10:10 +0000 Subject: [PATCH 2/2] x --- .github/tsan-suppressions.txt | 22 +++ .github/workflows/basic.yml | 6 + sycl/sycl-rs/src/lib.rs | 5 + sycl/sycl-rs/src/usmbox.rs | 254 ++++++++++++++++++++++++++++++++-- 4 files changed, 277 insertions(+), 10 deletions(-) create mode 100644 .github/tsan-suppressions.txt 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 2ec476e..713fba3 100644 --- a/.github/workflows/basic.yml +++ b/.github/workflows/basic.yml @@ -138,8 +138,14 @@ jobs: # 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 diff --git a/sycl/sycl-rs/src/lib.rs b/sycl/sycl-rs/src/lib.rs index 0073685..8907d32 100644 --- a/sycl/sycl-rs/src/lib.rs +++ b/sycl/sycl-rs/src/lib.rs @@ -106,6 +106,11 @@ //! [`Queue`](crate::queue::Queue) directly. To do the same asynchronously you have to `.await` an //! event returned by [`Queue::barrier()`](crate::queue::Queue::barrier). //! +//! Dropping an enqueued allocation or its future (including a losing `tokio::select!` branch) +//! blocks until initialization completes before freeing its memory. Other events and queues do not +//! wait on drop. If an initialization wait fails, the allocation is leaked to avoid freeing memory +//! still in use; explicit wait/await returns the error, while cancellation during drop panics. +//! //! All basic SYCL wrapper types (`Queue`, `Event`, `Context`, `Platform`, `Device`) are thread safe as //! indicated by the provided [`Send`] and [`Sync`] trait implementations. However - `UsmBox`es are //! not thread-safe. If you need a thread-safe `UsmBox` you need to wrap it in an `Arc>`. diff --git a/sycl/sycl-rs/src/usmbox.rs b/sycl/sycl-rs/src/usmbox.rs index b9ba1ab..9dd21ed 100644 --- a/sycl/sycl-rs/src/usmbox.rs +++ b/sycl/sycl-rs/src/usmbox.rs @@ -109,24 +109,54 @@ pub type SharedUsmBox = UsmBox>; pub type DeviceUsmBox = UsmBox>; /// A [`UsmBox`] whose initialization has been enqueued. You need to wait/await it. +/// +/// Dropping an enqueued array blocks until initialization completes before freeing its memory. +/// If waiting fails, the allocation is leaked rather than freed while it may still be in use. +/// A failed wait during drop panics. pub struct EnqueuedUsmBox { - array: UsmBox, + array: Option>, event: Event, } impl EnqueuedUsmBox { pub(crate) fn new(array: UsmBox, event: Event) -> Self { - Self { array, event } + Self { + array: Some(array), + event, + } } -} -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. + /// If waiting fails, the allocation is leaked because completion cannot be guaranteed. pub fn wait(mut self) -> Result> { - self.event.wait().map(|_| self.array) + let result = self.event.wait(); + self.finish(result) + } + + fn finish(&mut self, result: Result<()>) -> Result> { + let array = self.array.take().expect("allocation already consumed"); + match result { + Ok(()) => Ok(array), + Err(error) => { + // A failed wait does not guarantee that the device has stopped using the memory. + std::mem::forget(array); + Err(error) + } + } + } +} + +impl Drop for EnqueuedUsmBox { + fn drop(&mut self) { + if self.array.is_some() { + let result = self.event.wait(); + drop( + self.finish(result) + .expect("failed to wait for USM initialization during drop; allocation leaked"), + ); + } } } @@ -136,8 +166,11 @@ pub type EnqueuedDeviceUsmBox = EnqueuedUsmBox { - array: Option>, + enqueued: EnqueuedUsmBox, #[pin] event_future: EventFuture, } @@ -148,7 +181,7 @@ impl Future for UsmBoxFuture { let this = self.project(); this.event_future .poll(cx) - .map(|result| result.map(|_| this.array.take().unwrap())) + .map(|result| this.enqueued.finish(result)) } } @@ -158,8 +191,8 @@ impl IntoFuture for EnqueuedUsmBox { fn into_future(self) -> Self::IntoFuture { Self::IntoFuture { - array: Some(self.array), - event_future: self.event.into_future(), + event_future: self.event.clone().into_future(), + enqueued: self, } } } @@ -185,3 +218,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) + } +}