use super::graph::{Backend, NodeState, Operation, Prepared, Slot, Use};
use super::*;
#[cfg(feature = "npu")]
use crate::benchmark::percentile;
use std::{
sync::{
atomic::{AtomicUsize, Ordering},
mpsc,
},
time::Duration,
};
#[test]
#[ignore = "requires native GPU"]
fn allocation_stream_does_not_retain_released_tracked_storage() -> Result<()> {
let manager = crate::residency::ResidencyManager::new(1024)?;
let runtime = Runtime::with_options(RuntimeOptions {
memory_budget: Some(manager.budget()),
..Default::default()
})?;
let buffer = runtime.allocate(1024, MemoryPlacement::GpuLocal)?;
let native = buffer.storage.native.as_ref().unwrap().clone();
assert_eq!(manager.statistics().reserved_bytes, 1024);
drop(buffer);
assert_eq!(manager.statistics().reserved_bytes, 0);
assert!(
native.exclusively_owned(),
"released storage remains on the allocation stream"
);
Ok(())
}
#[test]
fn mapped_views_limit_bytes_but_reserve_the_whole_allocation() {
let runtime = runtime_without_workers();
let root = buffer(&runtime);
let view = root.slice(7..19).unwrap();
assert!(view.is_host_visible());
let mut write = view.map_write().unwrap();
write.fill(42);
assert!(matches!(root.try_map_read(), Err(Error::Busy(_))));
assert!(matches!(
root.slice(32..48).unwrap().map_write(),
Err(Error::Busy(_))
));
drop(write);
let read = view.map_read().unwrap();
assert_eq!(&*read, &[42; 12]);
assert!(matches!(root.try_map_write(), Err(Error::Busy(_))));
drop(read);
let read = root.map_read().unwrap();
assert_eq!(&read[..7], &[0; 7]);
assert_eq!(&read[7..19], &[42; 12]);
assert!(read[19..].iter().all(|&v| v == 0));
}
pub(super) fn buffer(runtime: &Runtime) -> Buffer {
#[allow(clippy::arc_with_non_send_sync)]
let memory = Arc::new(std::cell::UnsafeCell::new([0u8; 256]));
Buffer {
storage: Arc::new(Storage {
native: None,
accounted: false,
gpu: None,
pointer: memory.get().cast::<u8>(),
bytes: 256,
shared: true,
host: Mutex::new(HostState::default()),
visibility: Mutex::new(Visibility::new()),
runtime: runtime.inner.clone(),
_test_memory: Some(memory),
_reservation: None,
}),
}
}
#[test]
fn interval_frontiers_preserve_reference_order_and_compact_repeated_writes() {
use std::collections::BTreeSet;
let runtime = Runtime::new().unwrap();
let mut root = buffer(&runtime);
Arc::get_mut(&mut root.storage).unwrap().shared = false;
let mut random = 1234567u64;
for _ in 0..if cfg!(miri) { 2 } else { 64 } {
let mut frontier = dependencies::Frontier::default();
let mut history: Vec<Vec<Use>> = Vec::new();
let mut ancestors: Vec<BTreeSet<usize>> = Vec::new();
for node in 0..if cfg!(miri) { 16 } else { 64 } {
let mut uses = Vec::new();
for _ in 0..2 {
random = random.wrapping_mul(6364136223846793005).wrapping_add(1);
let start = ((random >> 32) as usize % 16) * 8;
let length = ((random >> 40) as usize % 16 + 1) * 8;
let access =
[Access::Read, Access::Write, Access::ReadWrite][(random >> 48) as usize % 3];
uses.push(Use {
view: root.slice(start..start + length).unwrap(),
access,
});
}
let mut reach = BTreeSet::new();
for previous in frontier.dependencies(node, &uses) {
assert!(previous < node);
reach.insert(previous);
reach.extend(&ancestors[previous]);
}
for (previous, old) in history.iter().enumerate() {
if old.iter().any(|a| uses.iter().any(|b| a.conflicts(b))) {
assert!(
reach.contains(&previous),
"lost conflict {previous} -> {node}"
);
}
}
let summary = dependencies::summarize(uses.clone());
for old in &history {
assert_eq!(
dependencies::conflicts(&dependencies::summarize(old.clone()), &summary),
old.iter().any(|a| uses.iter().any(|b| a.conflicts(b)))
);
}
ancestors.push(reach);
history.push(uses);
}
}
let mut frontier = dependencies::Frontier::default();
let uses = vec![Use {
view: root.view(),
access: Access::Write,
}];
for node in 0..2048 {
assert_eq!(
frontier.dependencies(node, &uses),
if node == 0 { vec![] } else { vec![node - 1] }
);
}
assert_eq!(
dependencies::summarize((0..2048).flat_map(|_| uses.clone())).len(),
1
);
}
pub(super) fn mock(
runtime: &Runtime,
engine: Engine,
buffer: &Buffer,
access: Access,
action: impl Fn() -> Result<()> + Send + Sync + 'static,
) -> ExecutableGraph {
let uses = vec![Use {
view: buffer.view(),
access,
}];
ExecutableGraph {
inner: Arc::new(Prepared {
quarantined: std::sync::atomic::AtomicBool::new(false),
signals: (0..runtime.options.graph_slots)
.map(|_| completion::Signal::new())
.collect(),
runtime: runtime.inner.clone(),
parallel: false,
operations: vec![Operation {
lane: GpuLane::Compute,
copy_bytes: 0,
backend: Backend::Mock(engine, Arc::new(action)),
uses: uses.clone(),
dependencies: vec![],
}],
uses,
slots: Mutex::new(
(0..runtime.options.graph_slots)
.map(|_| Slot {
nodes: vec![NodeState::Pending],
occupied: false,
failure: None,
})
.collect(),
),
}),
}
}
#[test]
fn native_stages_share_compute_but_leave_transfer_lanes_available() {
let runtime = Runtime::new().unwrap();
let compute = mock(
&runtime,
Engine::Gpu,
&buffer(&runtime),
Access::Write,
|| Ok(()),
);
let mut upload = mock(
&runtime,
Engine::Gpu,
&buffer(&runtime),
Access::Write,
|| Ok(()),
);
Arc::get_mut(&mut upload.inner).unwrap().operations[0].lane = GpuLane::Upload;
let mut native = unsafe { NativeSession::new(&runtime, (), |_| Ok(())) };
let completion = unsafe {
native.run(|_| {
let completion = compute.submit().unwrap();
upload.submit().unwrap().wait().unwrap();
assert!(
!completion.is_complete(),
"compute ran inside a native stage"
);
completion
})
}
.unwrap();
completion.wait().unwrap();
}
#[test]
fn native_stages_count_toward_submission_capacity() {
let runtime = Runtime::with_options(RuntimeOptions {
max_submissions: 1,
..Default::default()
})
.unwrap();
let graph = mock(
&runtime,
Engine::Gpu,
&buffer(&runtime),
Access::Write,
|| Ok(()),
);
let mut native = unsafe { NativeSession::new(&runtime, (), |_| Ok(())) };
unsafe {
native.run(|_| {
assert!(matches!(graph.submit(), Err(Error::Busy(_))));
let runtime = runtime.clone();
std::thread::spawn(move || {
let mut other = NativeSession::new(&runtime, (), |_| panic!("unadmitted fence"));
assert!(matches!(
other.run::<()>(|_| panic!("unadmitted stage")),
Err(Error::Busy(_))
));
})
.join()
.unwrap();
})
}
.unwrap();
graph.submit().unwrap().wait().unwrap();
}
#[test]
fn failed_dependency_does_not_poison_or_quarantine_untouched_output() {
let runtime = Runtime::new().unwrap();
let output = buffer(&runtime);
let signal = completion::Signal::new();
signal.reset();
signal.finish(Some(Arc::new(Error::DeviceLost("injected".into()))));
let failure = Completion::new(signal, Arc::downgrade(&runtime.inner.core));
let runs = Arc::new(AtomicUsize::new(0));
let seen = runs.clone();
let consumer = mock(&runtime, Engine::Gpu, &output, Access::Write, move || {
seen.fetch_add(1, Ordering::Relaxed);
Ok(())
});
let dependent = consumer
.submit_after(std::slice::from_ref(&failure))
.unwrap();
assert!(dependent.wait().is_err());
assert_eq!(runs.load(Ordering::Relaxed), 0);
assert!(!consumer.inner.quarantined.load(Ordering::Acquire));
drop(dependent);
consumer.submit().unwrap().wait().unwrap();
assert_eq!(runs.load(Ordering::Relaxed), 1);
assert!(output.map_read().is_ok());
}
pub(super) fn runtime_without_workers() -> Runtime {
let options = RuntimeOptions::default();
Runtime {
inner: Arc::new(RuntimeOwner {
core: Arc::new(Core::new(options.max_submissions)),
workers: Mutex::new(Vec::new()),
gpu_index: options.gpu_index,
copy_streams: std::array::from_fn(|_| Mutex::new(None)),
allocation_stream: Mutex::new(None),
}),
options,
}
}
#[test]
fn independent_gpu_lanes_can_run_together() {
let runtime = Runtime::new().unwrap();
let (started, received) = mpsc::channel();
let mut releases = Vec::new();
let mut completions = Vec::new();
for lane in [GpuLane::Upload, GpuLane::Compute, GpuLane::Download] {
let storage = buffer(&runtime);
let (release, released) = mpsc::channel();
releases.push(release);
let released = Mutex::new(released);
let started = started.clone();
let mut graph = mock(&runtime, Engine::Gpu, &storage, Access::Write, move || {
started.send(lane).unwrap();
released
.lock()
.unwrap()
.recv_timeout(Duration::from_secs(5))
.map_err(|error| Error::Message(error.to_string()))?;
Ok(())
});
Arc::get_mut(&mut graph.inner).unwrap().operations[0].lane = lane;
completions.push(graph.submit().unwrap());
}
let observed = (0..3)
.map(|_| received.recv_timeout(Duration::from_secs(3)))
.collect::<Vec<_>>();
for release in releases {
let _ = release.send(());
}
for completion in completions {
completion.wait().unwrap();
}
assert!(
observed.iter().all(|result| result.is_ok()),
"lanes were serialized: {observed:?}"
);
}
#[test]
fn gpu_lane_changes_preserve_conflicting_submission_order() {
let runtime = Runtime::new().unwrap();
let storage = buffer(&runtime);
let sequence = Arc::new(AtomicUsize::new(0));
let mut completions = Vec::new();
for (index, lane) in [GpuLane::Upload, GpuLane::Compute, GpuLane::Download]
.into_iter()
.enumerate()
{
let sequence = sequence.clone();
let mut graph = mock(
&runtime,
Engine::Gpu,
&storage,
Access::ReadWrite,
move || {
assert_eq!(sequence.fetch_add(1, Ordering::SeqCst), index);
Ok(())
},
);
Arc::get_mut(&mut graph.inner).unwrap().operations[0].lane = lane;
completions.push(graph.submit().unwrap());
}
for completion in completions {
completion.wait().unwrap();
}
assert_eq!(sequence.load(Ordering::SeqCst), 3);
}
#[test]
fn writer_waits_for_all_readers_completed_out_of_order() {
let runtime = runtime_without_workers();
let buffer = buffer(&runtime);
let reads = mock(&runtime, Engine::Gpu, &buffer, Access::Read, || Ok(()));
let writes = mock(&runtime, Engine::Npu, &buffer, Access::Write, || Ok(()));
let first = reads.submit().unwrap();
let second = reads.submit().unwrap();
let writer = writes.submit().unwrap();
let core = &runtime.inner.core;
scheduler::help(core, &writer.signal);
assert!(!writer.is_complete());
second.wait().unwrap();
scheduler::help(core, &writer.signal);
assert!(!writer.is_complete());
first.wait().unwrap();
writer.wait().unwrap();
drop((first, second, writer));
writes.submit().unwrap().wait().unwrap();
}
#[test]
fn cancelling_a_blocked_job_preserves_other_dependencies() {
let runtime = runtime_without_workers();
let shared = buffer(&runtime);
let separate = buffer(&runtime);
let writes = mock(&runtime, Engine::Gpu, &shared, Access::Write, || Ok(()));
let cancelled = mock(&runtime, Engine::Npu, &shared, Access::Write, || {
panic!("cancelled work must not execute")
});
let reads = mock(&runtime, Engine::Npu, &shared, Access::Read, || Ok(()));
let independent = mock(&runtime, Engine::Npu, &separate, Access::Write, || Ok(()));
let first = writes.submit().unwrap();
let middle = cancelled.submit().unwrap();
let last = reads.submit().unwrap();
let other = independent.submit().unwrap();
let core = &runtime.inner.core;
scheduler::help(core, &last.signal);
assert!(!last.is_complete());
middle.cancel();
assert!(matches!(middle.wait(), Err(Error::Cancelled)));
scheduler::help(core, &last.signal);
assert!(!last.is_complete());
other.wait().unwrap();
first.wait().unwrap();
last.wait().unwrap();
}
#[test]
fn aliases_and_host_guards_share_one_access_domain() {
let runtime = Runtime::new().unwrap();
let buffer = buffer(&runtime);
let alias = buffer.clone();
let mut write = buffer.map_write().unwrap();
write[0] = 7;
assert!(matches!(alias.try_map_read(), Err(Error::Busy(_))));
assert!(matches!(alias.try_map_write(), Err(Error::Busy(_))));
let graph = mock(&runtime, Engine::Gpu, &alias, Access::Read, || Ok(()));
assert!(matches!(graph.submit(), Err(Error::Busy(_))));
drop(write);
let read = alias.map_read().unwrap();
assert_eq!(read[0], 7);
let second = buffer.map_read().unwrap();
assert_eq!(second[0], 7);
graph.submit().unwrap().wait().unwrap();
assert!(matches!(buffer.try_map_write(), Err(Error::Busy(_))));
}
#[test]
fn conflicts_order_across_submissions_and_detached_handles() {
let runtime = Runtime::new().unwrap();
let buffer = buffer(&runtime);
let value = Arc::new(AtomicUsize::new(0));
let (started_tx, started_rx) = mpsc::sync_channel(1);
let (release_tx, release_rx) = mpsc::sync_channel(1);
let release_rx = Mutex::new(release_rx);
let writer_value = value.clone();
let producer = mock(&runtime, Engine::Gpu, &buffer, Access::Write, move || {
started_tx.send(()).unwrap();
release_rx.lock().unwrap().recv().unwrap();
writer_value.store(42, Ordering::Release);
Ok(())
});
let reader_value = value.clone();
let consumer = mock(&runtime, Engine::Npu, &buffer, Access::Read, move || {
assert_eq!(reader_value.load(Ordering::Acquire), 42);
Ok(())
});
drop(producer.submit().unwrap());
started_rx.recv_timeout(Duration::from_secs(2)).unwrap();
let done = consumer.submit().unwrap();
assert!(!done.wait_timeout(Duration::from_millis(10)).unwrap());
assert!(matches!(buffer.try_map_read(), Err(Error::Busy(_))));
drop(producer);
release_tx.send(()).unwrap();
assert!(done.wait_timeout(Duration::from_secs(2)).unwrap());
}
#[test]
fn independent_engines_progress_while_another_device_waits() {
let runtime = Runtime::new().unwrap();
let a = buffer(&runtime);
let b = buffer(&runtime);
let (tx, rx) = mpsc::sync_channel(1);
let rx = Mutex::new(rx);
let gpu = mock(&runtime, Engine::Gpu, &a, Access::Write, move || {
rx.lock().unwrap().recv().unwrap();
Ok(())
});
let npu = mock(&runtime, Engine::Npu, &b, Access::Write, || Ok(()));
let first = gpu.submit().unwrap();
let second = npu.submit().unwrap();
assert!(second.wait_timeout(Duration::from_secs(2)).unwrap());
assert!(!first.is_complete());
tx.send(()).unwrap();
first.wait().unwrap();
}
#[test]
fn independent_regions_in_one_graph_progress_with_a_host_waiter() {
let runtime = Runtime::new().unwrap();
let a = buffer(&runtime);
let b = buffer(&runtime);
let (release_tx, release_rx) = mpsc::sync_channel(1);
let release_rx = Mutex::new(release_rx);
let (ready_tx, ready_rx) = mpsc::sync_channel(1);
let mut graph = mock(&runtime, Engine::Gpu, &a, Access::Write, move || {
release_rx.lock().unwrap().recv().unwrap();
Ok(())
});
let prepared = Arc::get_mut(&mut graph.inner).unwrap();
prepared.parallel = true;
let usage = Use {
view: b.view(),
access: Access::Write,
};
prepared.uses.push(usage.clone());
prepared.operations.push(Operation {
lane: GpuLane::Compute,
copy_bytes: 0,
dependencies: vec![],
uses: vec![usage],
backend: Backend::Mock(
Engine::Npu,
Arc::new(move || {
ready_tx.send(()).unwrap();
Ok(())
}),
),
});
for slot in prepared.slots.get_mut().unwrap() {
slot.nodes.push(NodeState::Pending);
}
let completion = graph.submit().unwrap();
std::thread::scope(|scope| {
let mapping = scope.spawn(|| {
let guard = a.map_read().unwrap();
assert_eq!(guard[0], 0);
});
let progressed = ready_rx.recv_timeout(Duration::from_secs(2));
release_tx.send(()).unwrap();
completion.wait().unwrap();
mapping.join().unwrap();
assert!(progressed.is_ok(), "independent NPU region was stalled");
});
}
#[test]
fn cancellation_drains_running_work_and_prevents_dependents() {
let runtime = Runtime::new().unwrap();
let buffer = buffer(&runtime);
let (tx, rx) = mpsc::sync_channel(1);
let rx = Mutex::new(rx);
let producer = mock(&runtime, Engine::Gpu, &buffer, Access::Write, move || {
rx.lock().unwrap().recv().unwrap();
Ok(())
});
let calls = Arc::new(AtomicUsize::new(0));
let count = calls.clone();
let consumer = mock(&runtime, Engine::Npu, &buffer, Access::Read, move || {
count.fetch_add(1, Ordering::Relaxed);
Ok(())
});
let first = producer.submit().unwrap();
let second = consumer.submit().unwrap();
second.cancel();
tx.send(()).unwrap();
first.wait().unwrap();
assert!(matches!(second.wait(), Err(Error::Cancelled)));
assert_eq!(calls.load(Ordering::Relaxed), 0);
}
#[test]
fn retained_observers_apply_bounded_backpressure() {
let runtime = Runtime::with_options(RuntimeOptions {
graph_slots: 1,
..Default::default()
})
.unwrap();
let buffer = buffer(&runtime);
let graph = mock(&runtime, Engine::Gpu, &buffer, Access::Read, || Ok(()));
let done = graph.submit().unwrap();
done.wait().unwrap();
assert!(matches!(graph.submit(), Err(Error::Busy(_))));
}
#[test]
fn view_bounds_and_writable_alias_contracts_are_rejected() {
let runtime = Runtime::new().unwrap();
let buffer = buffer(&runtime);
assert!(buffer.slice(0..257).is_err());
assert!(buffer.slice(128..256).unwrap().slice(0..129).is_err());
let binding = BindingContract {
bytes: 128,
alignment: 4,
access: Access::ReadWrite,
layout: "bytes".into(),
};
let contract = KernelContract {
bindings: vec![binding.clone(), binding],
constants: vec![],
};
assert!(
contract
.check(&[buffer.view(), buffer.slice(0..128).unwrap()])
.is_err()
);
assert!(
contract
.check(&[
buffer.slice(0..128).unwrap(),
buffer.slice(128..256).unwrap()
])
.is_ok()
);
}
#[test]
fn a_completed_single_slot_can_be_reused_immediately() {
let runtime = Runtime::with_options(RuntimeOptions {
graph_slots: 1,
..Default::default()
})
.unwrap();
let buffer = buffer(&runtime);
let graph = mock(&runtime, Engine::Gpu, &buffer, Access::Read, || Ok(()));
for _ in 0..256 {
graph.submit().unwrap().wait().unwrap();
}
}
#[test]
#[cfg_attr(
miri,
ignore = "intentionally quarantines an uncertain native allocation"
)]
fn uncertain_failure_poison_is_shared_and_resources_are_quarantined() {
let runtime = Runtime::new().unwrap();
let manager = crate::residency::ResidencyManager::new(256).unwrap();
let mut buffer = buffer(&runtime);
Arc::get_mut(&mut buffer.storage).unwrap()._reservation =
Some(manager.budget().reserve(256).unwrap());
let weak = Arc::downgrade(&buffer.storage);
let alias = buffer.clone();
let graph = mock(&runtime, Engine::Gpu, &buffer, Access::Write, || {
Err(crate::Error::Backend {
backend: "test",
operation: "partial submit",
code: -2,
message: "injected abort failure".into(),
})
});
let completion = graph.submit().unwrap();
assert!(matches!(
completion.wait(),
Err(crate::Error::Execution { .. })
));
assert!(matches!(
alias.try_map_read(),
Err(crate::Error::DeviceLost(_))
));
assert!(matches!(graph.submit(), Err(crate::Error::DeviceLost(_))));
drop(completion);
drop(graph);
drop(alias);
drop(buffer);
drop(runtime);
assert!(weak.upgrade().is_some());
assert_eq!(manager.statistics().reserved_bytes, 256);
assert!(manager.budget().reserve(1).is_err());
}
#[test]
#[cfg(feature = "npu")]
#[ignore = "requires XDNA2, shared ABI 1 and HRX_TEST_NPU_DIR; prints paired backend baseline"]
fn heterogeneous_latency_against_direct_backend() -> Result<()> {
use std::time::Instant;
let runtime = Runtime::new()?;
let program = runtime.npu(0)?;
let artifact = crate::loom::Compiler::for_target(None, program.target())?
.module(include_str!("../../tests/kernels/copy.xdna.loom"))
.compile(&crate::loom::Specialization::new("copy").with_config("copy.packets", "256"))?;
let bytes = 1 << 20;
let binding = |bytes, access| BindingContract {
bytes,
alignment: 4,
access,
layout: "passthrough".into(),
};
let kernel = unsafe {
program.load_artifact(
&artifact,
1,
KernelContract {
bindings: vec![binding(bytes, Access::Read), binding(bytes, Access::Write)],
constants: vec![],
},
)
}?;
let a = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
let c = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
let mut graph = runtime.graph();
graph.fill(a.view(), 0x3d)?;
graph.npu(&kernel, &[a.view(), c.view()])?;
let graph = graph.prepare()?;
let direct = || -> Result<()> {
for operation in &graph.inner.operations {
operation.run()?;
}
Ok(())
};
for _ in 0..10 {
graph.submit()?.wait()?;
direct()?;
}
let mut scheduled = Vec::with_capacity(100);
let mut baseline = Vec::with_capacity(100);
let before = runtime.statistics();
for iteration in 0..100 {
if iteration % 2 == 0 {
let start = Instant::now();
direct()?;
baseline.push(start.elapsed().as_secs_f64());
let start = Instant::now();
graph.submit()?.wait()?;
scheduled.push(start.elapsed().as_secs_f64());
} else {
let start = Instant::now();
graph.submit()?.wait()?;
scheduled.push(start.elapsed().as_secs_f64());
let start = Instant::now();
direct()?;
baseline.push(start.elapsed().as_secs_f64());
}
}
baseline.sort_by(f64::total_cmp);
scheduled.sort_by(f64::total_cmp);
let after = runtime.statistics();
assert_eq!(before.allocations, after.allocations);
assert_eq!(before.imports, after.imports);
assert_eq!(after.copied_bytes, 0);
assert!(c.map_read()?.iter().all(|&b| b == 0x3d));
println!(
"paired_direct_p50_us={:.3} scheduled_p50_us={:.3} ratio={:.3} scheduled_p95_us={:.3}",
percentile(&baseline, 50) * 1e6,
percentile(&scheduled, 50) * 1e6,
percentile(&scheduled, 50) / percentile(&baseline, 50),
percentile(&scheduled, 95) * 1e6
);
Ok(())
}
#[test]
fn host_sync_releases_scheduler_lock_and_rolls_back_failed_leases() {
let runtime = runtime_without_workers();
let buffer = buffer(&runtime);
for write in [false, true] {
let result = buffer.acquire_with(write, false, || {
assert!(runtime.inner.core.state.try_lock().is_ok());
let graph = mock(&runtime, Engine::Gpu, &buffer, Access::Write, || Ok(()));
assert!(matches!(graph.submit(), Err(Error::Busy(_))));
Err(Error::Message("injected sync failure".into()))
});
assert!(result.is_err());
drop(buffer.try_map_write().unwrap());
}
}
#[test]
fn cache_transitions_flush_host_writes_for_both_devices() {
for (producer, consumer, to_device) in [
(Engine::Host, Engine::Gpu, true),
(Engine::Host, Engine::Npu, true),
(Engine::Gpu, Engine::Host, false),
(Engine::Gpu, Engine::Npu, true),
(Engine::Npu, Engine::Host, false),
(Engine::Npu, Engine::Gpu, false),
] {
let mut visibility = Visibility::new();
visibility.wrote(producer);
assert!(!visibility.visible[consumer as usize]);
assert_eq!(visibility.sync_to_device(consumer), to_device);
}
}
#[test]
fn completion_releases_submission_owners_before_waking_observers() {
use std::{
future::Future,
pin::Pin,
sync::atomic::AtomicBool,
task::{Context, Wake, Waker},
};
struct Owner(Arc<AtomicBool>);
impl Drop for Owner {
fn drop(&mut self) {
self.0.store(true, Ordering::Release);
}
}
struct Observer {
released: Arc<AtomicBool>,
observed: AtomicBool,
}
impl Wake for Observer {
fn wake(self: Arc<Self>) {
self.observed
.store(self.released.load(Ordering::Acquire), Ordering::Release);
}
}
let runtime = runtime_without_workers();
let released = Arc::new(AtomicBool::new(false));
let owner = Owner(released.clone());
let graph = mock(
&runtime,
Engine::Gpu,
&buffer(&runtime),
Access::Read,
move || {
let _retained = &owner;
Ok(())
},
);
let mut completion = graph.submit().unwrap();
drop(graph);
let observer = Arc::new(Observer {
released,
observed: AtomicBool::new(false),
});
let waker = Waker::from(observer.clone());
assert!(
Pin::new(&mut completion)
.poll(&mut Context::from_waker(&waker))
.is_pending()
);
completion.wait().unwrap();
assert!(
observer.observed.load(Ordering::Acquire),
"completion woke its observer before releasing the submission owner"
);
}
#[test]
fn completion_waker_can_drop_its_observer_and_reuse_the_only_graph_slot() {
use std::{
future::Future,
pin::Pin,
task::{Context, Wake, Waker},
};
struct Resubmit {
graph: ExecutableGraph,
prior: Mutex<Option<Completion>>,
next: Mutex<Option<Result<Completion>>>,
}
impl Wake for Resubmit {
fn wake(self: Arc<Self>) {
drop(self.prior.lock().unwrap().take());
*self.next.lock().unwrap() = Some(self.graph.submit());
}
}
let mut runtime = runtime_without_workers();
runtime.options.graph_slots = 1;
runtime.inner.core.state.lock().unwrap().capacity = 1;
let graph = mock(
&runtime,
Engine::Gpu,
&buffer(&runtime),
Access::Read,
|| Ok(()),
);
let completion = graph.submit().unwrap();
let signal = completion.signal.clone();
let observer = Arc::new(Resubmit {
graph,
prior: Mutex::new(Some(completion)),
next: Mutex::new(None),
});
let waker = Waker::from(observer.clone());
assert!(
Pin::new(observer.prior.lock().unwrap().as_mut().unwrap())
.poll(&mut Context::from_waker(&waker))
.is_pending()
);
scheduler::help(&runtime.inner.core, &signal);
observer
.next
.lock()
.unwrap()
.take()
.unwrap()
.unwrap()
.wait()
.unwrap();
}