From 261058ca728623d6b3a13e65fc84b2ac743e405d Mon Sep 17 00:00:00 2001 From: Aryan Ravishankar Date: Thu, 27 Aug 2026 20:44:30 -0500 Subject: [PATCH] Pipeline the dispatch lane: depth-4 ring, per-job timeline retirement (#151, #149) The worker previously held the stream mutex across hrx_stream_dispatch plus a blocking hrx_stream_synchronize, one job at a time: one submission outstanding, zero overlap, and allocate_buffer queued behind every running dispatch. The lane now keeps up to four jobs in flight. Dispatch (record + flush + timeline position) holds the stream mutex briefly; the completion wait blocks on the stream's timeline semaphore in bounded slices with no lock held. Ring capacity is released in finish() before the terminal state becomes observable, so a caller that polls Complete and immediately resubmits never bounces off a stale count. Every fault-tier semantic maps one-to-one: a definite error still latches Failed and poisons; an untrusted boundary still leaves events pending and gates armed; on a tier-2 wedge the worker parks forever holding every in-flight job's retained resources and the stream, the same quarantine as before. Two on-metal findings shaped the implementation. The flushed batch's timeline value is assigned asynchronously: a position read immediately after the flush occasionally still reports the previous batch's target, and waiting on that retires early - observed as stale outputs (the max-pool NaN round reading the previous submission's bytes, roughly once per six cold runs). The stream is instance-private and dispatches are serialized, so each job's tick is the first value observed past its predecessor's; the position read now spins on that induction, watchdog-guarded. Second, throughput: pipelined and sequential submission measure identically (73-75 vs 69-75 microseconds per inference), and a batched-flush variant measured the same, so the per-submission floor is per-command driver/firmware round-trip cost inside one hardware context. docs/performance.md records the numbers; the honest conclusion is that further host-side submission restructuring cannot move the floor, and effective throughput comes from more work per dispatch (#151 steps 5-6) or parallel contexts (#121). The depth still pays for itself in semantics: multiple pending events (reference parity), submissions overlapping host-side polling and readback, and completion waits that no longer serialize allocate_buffer. New coverage, all on metal: a pipelined-payload test keeps four distinct submissions in flight for eight rounds and verifies every completion carries exactly its own payload; a deterministic ring-capacity test fills all four event slots, proves Busy at five, and proves reclaim restores exactly one; the pipelined-throughput benchmark reports amortized per-inference cost with oracle-validated warmups. The previously racy pending-release test now uses a new test-control HoldDispatch fault (a bounded delay before one healthy dispatch) instead of hoping the release beats real completion - pipelining made that race losable, observed twice in cold runs. hrx_stream_synchronize is no longer referenced and leaves the FFI; hrx_stream_flush, hrx_stream_get_timeline_position, and hrx_semaphore_wait join it, transcribed from the pinned headers. Co-Authored-By: Claude Opus 5 --- crates/virtio-accel-xdna/README.md | 7 +- crates/virtio-accel-xdna/SAFETY.md | 39 +- crates/virtio-accel-xdna/src/ffi.rs | 21 +- crates/virtio-accel-xdna/src/native.rs | 326 ++++++++++--- crates/virtio-accel-xdna/tests/hardware.rs | 539 ++++++++++++++++++++- docs/performance.md | 14 + 6 files changed, 827 insertions(+), 119 deletions(-) diff --git a/crates/virtio-accel-xdna/README.md b/crates/virtio-accel-xdna/README.md index 330bd8e..d23c11b 100644 --- a/crates/virtio-accel-xdna/README.md +++ b/crates/virtio-accel-xdna/README.md @@ -11,7 +11,8 @@ build time; a compile-only unsupported-runtime placeholder elsewhere. In a `va_xdna` build it runs the full `Accelerator` lifecycle — device/stream owner, `hrx_buffer` primitives (persistent mapping, range flush/invalidate, release), and a serialized dispatch worker bridging -`hrx_stream_dispatch`/`synchronize` to a latched nonblocking `poll_event`. `load_program` accepts +`hrx_stream_dispatch`/timeline-semaphore completion to a latched nonblocking `poll_event`, with +up to four submissions in flight per instance. `load_program` accepts the crate-local precompiled artifact format directly, and a TOSA artifact by admitting it and compiling it with the bounded aiecc helper subprocess (`compiler/xdna_compile.py`, run under the pinned toolchain venv in a cleared environment, content-addressed in a cache). The compilable TOSA @@ -104,10 +105,10 @@ RESCALE applies its signed INT8 output zero point only after exact 64-bit multip ## Completion and fault model One worker serializes each instance's accepted submissions. Finite timeouts are rejected before -admission because HRX exposes no cancellation primitive. A definite dispatch/synchronize failure +admission because HRX exposes no cancellation primitive. A definite dispatch or completion-wait failure becomes a stable terminal `Failed` event and poisons that backend instance (device-loss tier 1); the event and its buffers can still be released normally. A 120-second userspace watchdog, longer -than the kernel's 60-second NPU TDR, detects a synchronize call that never returns (tier 2). In that +than the kernel's 60-second NPU TDR, detects a dispatch or completion wait that never finishes (tier 2). In that case `poll_event` reports `DeviceLost`, the event remains pending and cannot be released, and the host must discard the backend instance. The detached worker retains the stream, executable, and buffer allocations so discarding cannot free native memory that HRX might still touch. diff --git a/crates/virtio-accel-xdna/SAFETY.md b/crates/virtio-accel-xdna/SAFETY.md index 942f8ee..95a62bc 100644 --- a/crates/virtio-accel-xdna/SAFETY.md +++ b/crates/virtio-accel-xdna/SAFETY.md @@ -68,9 +68,21 @@ the buffer's `TRANSFER_DESTINATION`/`TRANSFER_SOURCE` usage. ## Concurrency and the dispatch worker The HRX stream is not safe for concurrent use, so all stream access is serialized by the `Lane` -stream mutex: `allocate_buffer` locks it briefly, and the worker holds it across -`hrx_stream_dispatch` + `hrx_stream_synchronize`. `Stream` is `unsafe impl Send` (moved to the -worker, only ever dereferenced under that mutex). +stream mutex: `allocate_buffer` locks it briefly, and the worker locks it briefly per dispatch +(`hrx_stream_dispatch` + `hrx_stream_flush` + the timeline-position read). The completion wait +holds no lock at all: it blocks on the stream's timeline semaphore (`hrx_semaphore_wait` in +bounded slices), a standalone synchronization object that is valid while the lane retains the +stream and safe to wait on while another thread holds the stream mutex — so `allocate_buffer` +never queues behind a running dispatch. `Stream` is `unsafe impl Send` (moved to the worker, only +ever dereferenced under that mutex). + +Up to the ring depth (four) submissions are in flight on the stream at once; the stream executes +them in order and the worker retires them oldest-first. Each job's completion tick is the first +timeline value observed past its predecessor's: the flushed batch's value is assigned +asynchronously, so a position read on an unlucky schedule still reports the previous batch's +target, and waiting on that retires early (observed on metal as stale outputs before the +induction-anchored spin was added). The stream is instance-private and dispatches are serialized +under the mutex, which is what makes the induction sound. `submit` validates the bindings in one pass (a nonempty, within-limit list; a queue, program, and every buffer from one context; unique slots via a 256-bit occupancy mask; per-slot access; ranges; @@ -88,18 +100,23 @@ dereferenced only on the worker while the stream mutex is held. The in-flight ga those `BufferInner` Arcs, not caller structs: the contract requires the caller keep handles *alive* until the event is terminal, not address-stable, so a caller may legally move an `XdnaBuffer` mid-flight; the shared allocation keeps both handle and gate valid regardless. The worker, per job, -dispatches, synchronizes, +dispatches, waits for the job's timeline tick, `invalidate_range`s each output, clears the `in_flight` gates, and latches the event's terminal -state exactly once. While a buffer's gate is set, `write_buffer`/`read_buffer`/`free_buffer` reject +state exactly once — releasing the job's ring capacity before the terminal state becomes +observable, so a caller that polls completion and immediately resubmits never bounces off a stale +count. While a buffer's gate is set, `write_buffer`/`read_buffer`/`free_buffer` reject with `Busy`, so no host access or release races the device. -Finite timeouts are rejected before admission (no cancellation exists at any layer). A synchronize +Finite timeouts are rejected before admission (no cancellation exists at any layer). A dispatch or wait error latches the event `Failed` (a normal terminal state — the kernel TDR has quiesced the device) -and then poisons the instance, which refuses further work with `DeviceLost`. `poll_event` reads the -latched atomic state without touching HRX. The worker arms a 120-second watchdog immediately before -dispatch, longer than the kernel's 60-second NPU TDR. If HRX still has not returned, the watchdog -poisons the lane but deliberately leaves the accepted event pending and every gate armed: there is -no trustworthy completion boundary. `poll_event` then reports `DeviceLost`; event release remains +and then poisons the instance, which refuses further work with `DeviceLost`; jobs already accepted +behind the failure latch `Failed(DeviceLost)` without touching the dead stream. `poll_event` reads +the latched atomic state without touching HRX. The worker arms a 120-second watchdog around each +dispatch and each completion wait, longer than the kernel's 60-second NPU TDR. If the boundary is +declared lost, the watchdog poisons the lane but deliberately leaves the accepted events pending +and every gate armed: there is no trustworthy completion boundary. The worker then parks forever +holding every in-flight job's retained resources and the stream — the quarantine — and `Drop` +detaches it. `poll_event` then reports `DeviceLost`; event release remains retryably rejected as `Busy`; discarding the backend enters the quarantine described above. `EVENT_CANCELLATION` is not advertised. diff --git a/crates/virtio-accel-xdna/src/ffi.rs b/crates/virtio-accel-xdna/src/ffi.rs index 2abc976..27a7d77 100644 --- a/crates/virtio-accel-xdna/src/ffi.rs +++ b/crates/virtio-accel-xdna/src/ffi.rs @@ -71,6 +71,8 @@ pub(crate) type hrx_stream_t = *mut hrx_stream_s; pub(crate) type hrx_buffer_t = *mut hrx_buffer_s; /// `hrx_executable_t` — refcounted executable handle. pub(crate) type hrx_executable_t = *mut hrx_executable_s; +pub(crate) enum hrx_semaphore_s {} +pub(crate) type hrx_semaphore_t = *mut hrx_semaphore_s; /// Borrowed byte span (`hrx_const_byte_span_t`). #[repr(C)] @@ -87,6 +89,14 @@ pub(crate) struct hrx_string_view_t { } /// Dispatch grid configuration (`hrx_dispatch_config_t`); the amdxdna path uses {1,1,1}/{1,1,1}/0. +/// `hrx_timeline_point_t` (hrx_runtime.h:250): a stream-timeline completion marker. +#[repr(C)] +#[derive(Clone, Copy)] +pub(crate) struct hrx_timeline_point_t { + pub(crate) semaphore: hrx_semaphore_t, + pub(crate) value: u64, +} + #[repr(C)] pub(crate) struct hrx_dispatch_config_t { pub workgroup_count: [u32; 3], @@ -222,5 +232,14 @@ unsafe extern "C" { binding_count: usize, flags: u32, ) -> hrx_status_t; - pub(crate) fn hrx_stream_synchronize(stream: hrx_stream_t) -> hrx_status_t; + pub(crate) fn hrx_stream_flush(stream: hrx_stream_t) -> hrx_status_t; + pub(crate) fn hrx_stream_get_timeline_position( + stream: hrx_stream_t, + position: *mut hrx_timeline_point_t, + ) -> hrx_status_t; + pub(crate) fn hrx_semaphore_wait( + semaphore: hrx_semaphore_t, + value: u64, + timeout_ns: u64, + ) -> hrx_status_t; } diff --git a/crates/virtio-accel-xdna/src/native.rs b/crates/virtio-accel-xdna/src/native.rs index 003e9a9..ad3f606 100644 --- a/crates/virtio-accel-xdna/src/native.rs +++ b/crates/virtio-accel-xdna/src/native.rs @@ -41,8 +41,14 @@ use crate::{InitError, XDNA_ERROR_DOMAIN}; /// Upper bound advertised for a loaded artifact (mirrors the OpenVINO backend). const MAX_TOSA_ARTIFACT_BYTES: u64 = 256 * 1024 * 1024; -/// Default submission-ring depth (issue #85: one admitted request, matching the Hexagon backend). -const DEFAULT_RING_DEPTH: usize = 1; +/// Default submission-ring depth: up to four requests outstanding per instance. Issue #85 started +/// at one (matching the Hexagon backend); pipelined dispatch (#151/#149) raised it so consecutive +/// submissions overlap on the stream and the per-submission host cost amortizes. +const DEFAULT_RING_DEPTH: usize = 4; + +/// Completion waits poll the stream timeline in bounded slices so the worker observes a watchdog +/// wedge verdict promptly and no HRX call blocks unboundedly. +const WAIT_SLICE_NS: u64 = 100_000_000; /// AIE DMA descriptors transfer whole four-byte words, so a bound range must begin on a word /// boundary as well as cover its slot's exact byte length. The base mapping is page-aligned, so the @@ -85,6 +91,12 @@ fn check(status: ffi::hrx_status_t) -> Result<(), BackendError> { Err(backend_error_from_code(code)) } +/// The two `hrx_status_code_t` values the retire loop branches on by value. +mod code { + pub(super) const OK: super::ffi::hrx_status_code_t = 0; + pub(super) const DEADLINE_EXCEEDED: super::ffi::hrx_status_code_t = 4; +} + /// Map an HRX status code (`hrx_status_code_t`, mirroring IREE) to a `BackendError`. fn backend_error_from_code(code: ffi::hrx_status_code_t) -> BackendError { match code { @@ -250,9 +262,13 @@ struct Job { // the worker transfers that exclusive access. Everything else in the job (`Arc`s) is `Send`. unsafe impl Send for Job {} -/// The bounded submission ring. +/// The bounded submission ring. `queue` holds accepted jobs the worker has not yet dispatched; +/// `dispatched` counts jobs the worker has moved onto the device but not yet retired. Their sum is +/// bounded by the lane depth, so `submit` rejects with `Busy` exactly when `depth` submissions are +/// outstanding in any mix of queued and in-flight. struct Ring { queue: VecDeque, + dispatched: usize, stopping: bool, } @@ -268,6 +284,7 @@ struct WatchdogState { enum InjectedFault { Tier1, Tier2 { stall: Duration }, + HoldDispatch { hold: Duration }, } /// One-shot fault used by the on-metal fault-path tests. @@ -279,6 +296,9 @@ pub enum XdnaTestFault { Tier1, /// Hold the worker beyond the watchdog deadline without touching HRX. Tier2 { stall: Duration }, + /// Delay the worker before one normal dispatch (no fault): a deterministic pending window for + /// tests that assert in-flight semantics, which are otherwise a race against real completion. + HoldDispatch { hold: Duration }, } /// Test-only construction parameters. This API is absent unless `test-control` is enabled. @@ -360,44 +380,109 @@ impl Lane { } } - /// Run the worker loop: drain the ring one job at a time, dispatching and synchronizing under - /// the stream mutex, until stopped and empty. + /// Run the worker loop: keep up to `depth` jobs in flight on the stream, retiring the oldest + /// while later submissions are already recorded and executing. Dispatch (record + flush + + /// timeline position) holds the stream mutex briefly; the completion wait blocks on the + /// stream's timeline semaphore with no lock held, so `allocate_buffer` never queues behind a + /// running dispatch. In-flight jobs are worker-owned: on a tier-2 wedge the worker parks + /// forever holding them (and the `Arc` stream), forming the quarantine. fn run_worker(self: &Arc) { + let mut in_flight: VecDeque<(Job, ffi::hrx_timeline_point_t)> = VecDeque::new(); + // The timeline target of the most recently dispatched job. The stream is instance-private + // and dispatches are serialized under the stream mutex, so each job's target is exactly + // the first timeline value observed past its predecessor's (see `dispatch_job`). + let mut last_target: u64 = 0; loop { - let job = { - let mut ring = self.ring.lock().expect("ring mutex"); - loop { - if let Some(job) = ring.queue.pop_front() { - break job; + // Fill: move queued jobs onto the device up to the lane depth. + loop { + let job = { + let mut ring = self.ring.lock().expect("ring mutex"); + if in_flight.len() >= self.depth { + None + } else if let Some(job) = ring.queue.pop_front() { + ring.dispatched += 1; + Some(job) + } else { + None } - if ring.stopping { - return; + }; + let Some(job) = job else { break }; + match self.dispatch_job(job, &mut last_target) { + DispatchOutcome::InFlight(entry) => in_flight.push_back(entry), + // Terminal at dispatch (error, poisoned short-circuit, or an injected + // fault). `finish` released the ring capacity where a terminal state was + // latched; the deliberately-pending outcomes (untrusted boundary, injected + // tier 2) keep their slot and capacity, as quarantined work must. + DispatchOutcome::Retired => {} + } + } + // Retire the oldest in-flight job, or sleep until there is work. + if let Some((job, position)) = in_flight.pop_front() { + if !self.retire(job, position) { + // Tier-2 wedge: no trustworthy completion boundary exists for this job or + // anything recorded behind it. Park forever holding every in-flight job's + // retained resources and the stream; `Drop` detaches this thread. + loop { + std::thread::sleep(Duration::from_secs(3600)); } - ring = self.signal.wait(ring).expect("ring condvar"); } - }; - self.execute(job); + continue; + } + let mut ring = self.ring.lock().expect("ring mutex"); + loop { + if !ring.queue.is_empty() { + break; + } + if ring.stopping { + return; + } + ring = self.signal.wait(ring).expect("ring condvar"); + } } } - /// Dispatch one job on the stream, block on synchronize, make outputs host-visible, clear the - /// in-flight gates, and latch the event. A synchronize error latches `Failed` and poisons the - /// instance (device-loss tier 1). - fn execute(&self, job: Job) { + /// Dispatch one job: record it on the stream, flush, and capture its timeline position. On a + /// definite error the job is latched `Failed` here and the instance poisons (device-loss + /// tier 1); on a poisoned lane the job short-circuits to `Failed(DeviceLost)` without touching + /// the stream. Injected test faults are consumed here, before any HRX call. + fn dispatch_job(&self, job: Job, last_target: &mut u64) -> DispatchOutcome { #[cfg(feature = "test-control")] - let injected = self - .injected_fault - .lock() - .expect("fault-injector mutex") - .take(); - #[cfg(feature = "test-control")] - if let Some(InjectedFault::Tier2 { stall }) = injected { - let watchdog_generation = self.arm_watchdog(); - std::thread::sleep(stall); - let _ = self.disarm_watchdog(watchdog_generation); - // Tier 2 has no trustworthy completion boundary: leave the event pending and every - // gate armed. Discarding the backend quarantines this job's native resources. - return; + { + let injected = self + .injected_fault + .lock() + .expect("fault-injector mutex") + .take(); + match injected { + Some(InjectedFault::Tier2 { stall }) => { + let watchdog_generation = self.arm_watchdog(); + std::thread::sleep(stall); + let _ = self.disarm_watchdog(watchdog_generation); + // Tier 2 has no trustworthy completion boundary: leave the event pending and + // every gate armed. Discarding the backend quarantines this job's native + // resources. (No HRX call was made, so dropping the job itself is safe.) + return DispatchOutcome::Retired; + } + Some(InjectedFault::Tier1) => { + let generation = self.arm_watchdog(); + let trusted = self.disarm_watchdog(generation); + if trusted { + self.finish(job, Err(BackendError::DeviceLost)); + } + return DispatchOutcome::Retired; + } + Some(InjectedFault::HoldDispatch { hold }) => { + // A deterministic pending window: the event and every in-flight gate stay + // armed for at least `hold`, then the job proceeds normally. + std::thread::sleep(hold); + } + None => {} + } + } + // A poisoned lane refuses the stream: jobs accepted before the poison latch terminally. + if self.is_poisoned() { + self.finish(job, Err(BackendError::DeviceLost)); + return DispatchOutcome::Retired; } let config = ffi::hrx_dispatch_config_t { @@ -405,25 +490,106 @@ impl Lane { workgroup_size: [1, 1, 1], subgroup_size: 0, }; - #[cfg(feature = "test-control")] - let (trusted, result) = if matches!(injected, Some(InjectedFault::Tier1)) { - let generation = self.arm_watchdog(); - ( - self.disarm_watchdog(generation), - Err(BackendError::DeviceLost), + let program = job.program.as_ref(); + let stream = self.stream.lock().expect("stream mutex"); + // The watchdog covers the record/flush ioctls too: a driver hang here is the same + // ownership-boundary loss as one during the completion wait. + let watchdog_generation = self.arm_watchdog(); + // SAFETY: the stream, executable, and every bound buffer are retained by the lane/job for + // this call; the bindings slice is valid for `binding_count`; the config is a valid local; + // constants are unused on this path. + let dispatch = unsafe { + ffi::hrx_stream_dispatch( + stream.0.as_ptr(), + program.executable.as_ptr(), + program.ordinal, + &config, + ptr::null(), + 0, + job.bindings.as_ptr(), + job.bindings.len(), + 0, ) - } else { - self.dispatch_and_synchronize(&job, &config) }; - #[cfg(not(feature = "test-control"))] - let (trusted, result) = self.dispatch_and_synchronize(&job, &config); - + let result = check(dispatch).and_then(|()| { + // SAFETY: the stream is live; flush submits the recorded work without waiting. + check(unsafe { ffi::hrx_stream_flush(stream.0.as_ptr()) })?; + // The flushed batch's timeline value is assigned asynchronously: a position read + // before the dispatch (or immediately after the flush, on an unlucky schedule) still + // reports the previous batch's target, and waiting on that retires this job early -- + // observed on metal as stale outputs. The stream is instance-private and dispatches + // are serialized under this mutex, so only this flush can advance the timeline past + // the previous job's target: spin until it does, and that value is exactly this + // job's completion tick. The watchdog is armed, so a driver that never assigns is + // declared wedged rather than spun on forever. + loop { + let mut position = ffi::hrx_timeline_point_t { + semaphore: ptr::null_mut(), + value: 0, + }; + // SAFETY: the stream is live; the out-pointer is a valid local. + check(unsafe { + ffi::hrx_stream_get_timeline_position(stream.0.as_ptr(), &mut position) + })?; + if position.value > *last_target { + *last_target = position.value; + break Ok(position); + } + if self.wedged.load(Ordering::Acquire) { + break Err(BackendError::DeviceLost); + } + std::thread::yield_now(); + } + }); + drop(stream); + let trusted = self.disarm_watchdog(watchdog_generation); if !trusted { - // Synchronize returned only after the watchdog declared the ownership boundary lost. - // Keep the accepted event pending and do not publish the buffers as reusable. - return; + // The ioctl returned only after the watchdog declared the boundary lost: keep the + // event pending, keep the gates armed; the caller's teardown quarantines. + return DispatchOutcome::Retired; + } + match result { + Ok(position) => DispatchOutcome::InFlight((job, position)), + Err(error) => { + self.finish(job, Err(error)); + DispatchOutcome::Retired + } } + } + /// Wait for one in-flight job's timeline position and latch its terminal state. Returns + /// `false` when the watchdog declared the lane wedged while waiting (tier 2): the caller must + /// quarantine, because the device may still write through this and every later in-flight + /// job's bindings. + fn retire(&self, job: Job, position: ffi::hrx_timeline_point_t) -> bool { + let watchdog_generation = self.arm_watchdog(); + let result = loop { + // Bounded slices keep this wait off the stream mutex and let the watchdog's verdict + // surface promptly; the semaphore is the stream's own timeline object, valid while + // the lane retains the stream, and safe to wait on while another thread holds the + // stream mutex (it is a standalone synchronization object). + // SAFETY: `position.semaphore` is the live stream timeline semaphore (see above); the + // status is consumed. + let status = unsafe { + ffi::hrx_semaphore_wait(position.semaphore, position.value, WAIT_SLICE_NS) + }; + let code = unsafe { ffi::hrx_status_code(status) }; + unsafe { ffi::hrx_status_ignore(status) }; + match code { + code::OK => break Ok(()), + code::DEADLINE_EXCEEDED => { + if self.wedged.load(Ordering::Acquire) { + // Watchdog verdict while we sliced: same as disarm returning untrusted. + return false; + } + } + other => break Err(backend_error_from_code(other)), + } + }; + let trusted = self.disarm_watchdog(watchdog_generation); + if !trusted { + return false; + } let result = result.and_then(|()| { for &(buffer, offset, len) in &job.outputs { // SAFETY: each output buffer is live and persistently mapped; the range was @@ -432,8 +598,21 @@ impl Lane { } Ok(()) }); + self.finish(job, result); + true + } + /// Publish a job's terminal state: clear the in-flight gates, latch the event, and poison the + /// instance on failure (device-loss tier 1). + fn finish(&self, job: Job, result: Result<(), BackendError>) { let failed = result.is_err(); + // Release this job's ring capacity before the terminal state becomes observable: a caller + // that polls `Complete`, destroys the event, and immediately resubmits must never bounce + // off a stale `dispatched` count. + { + let mut ring = self.ring.lock().expect("ring mutex"); + ring.dispatched -= 1; + } // Clear the in-flight gates before publishing the terminal state, so a caller that observes // completion may immediately read or free the buffers. The retained `Arc`s remain valid // regardless of where (or whether) the caller's buffer values still live. @@ -450,37 +629,15 @@ impl Lane { self.poisoned.store(true, Ordering::Release); } } +} - fn dispatch_and_synchronize( - &self, - job: &Job, - config: &ffi::hrx_dispatch_config_t, - ) -> (bool, Result<(), BackendError>) { - let program = job.program.as_ref(); - let stream = self.stream.lock().expect("stream mutex"); - let watchdog_generation = self.arm_watchdog(); - // SAFETY: the stream, executable, and every bound buffer are retained by the lane/job for - // this call; the bindings slice is valid for `binding_count`; the config is a valid local; - // constants are unused on this path. - let dispatch = unsafe { - ffi::hrx_stream_dispatch( - stream.0.as_ptr(), - program.executable.as_ptr(), - program.ordinal, - config, - ptr::null(), - 0, - job.bindings.as_ptr(), - job.bindings.len(), - 0, - ) - }; - let result = check(dispatch).and_then(|()| { - // SAFETY: the stream is live; synchronize flushes and blocks until completion. - check(unsafe { ffi::hrx_stream_synchronize(stream.0.as_ptr()) }) - }); - (self.disarm_watchdog(watchdog_generation), result) - } +/// What became of one job handed to `dispatch_job`. +enum DispatchOutcome { + /// Recorded and flushed; retire it at this timeline position. + InFlight((Job, ffi::hrx_timeline_point_t)), + /// Terminal at dispatch: latched (error paths) or deliberately left pending (untrusted + /// boundary and the injected tier-2 fault). + Retired, } /// One HRX backend instance: a serialized dispatch lane over the shared device. @@ -529,6 +686,14 @@ impl XdnaAccelerator { InjectedFault::Tier2 { stall } } XdnaTestFault::Tier2 { .. } => return Err(InitError::Initialization), + // The hold must stay well inside the watchdog deadline: it delays a healthy + // dispatch, it does not simulate a hang. + XdnaTestFault::HoldDispatch { hold } + if !hold.is_zero() && hold * 2 <= config.watchdog_timeout => + { + InjectedFault::HoldDispatch { hold } + } + XdnaTestFault::HoldDispatch { .. } => return Err(InitError::Initialization), }; Self::initialize(config.watchdog_timeout, Some(fault)) } @@ -550,6 +715,7 @@ impl XdnaAccelerator { stream: Mutex::new(Stream(stream)), ring: Mutex::new(Ring { queue: VecDeque::with_capacity(depth), + dispatched: 0, stopping: false, }), signal: Condvar::new(), @@ -1314,7 +1480,7 @@ impl Accelerator for XdnaAccelerator { // Acceptance boundary: claim a ring entry and an event slot, then arm the in-flight gates. let slot = { let mut ring = self.lane.ring.lock().expect("ring mutex"); - if ring.queue.len() >= self.lane.depth { + if ring.queue.len() + ring.dispatched >= self.lane.depth { return reject(BackendError::Busy); } let Some(slot) = self.claim_slot() else { @@ -1400,9 +1566,9 @@ impl Accelerator for XdnaAccelerator { } impl Drop for XdnaEvent { - /// Reclaim the ring slot. `destroy_event` is the intended release path, but with a ring depth of - /// one a dropped-instead-of-destroyed event would strand the only slot and fail every later - /// submission with `Busy` for the life of the instance. + /// Reclaim the ring slot. `destroy_event` is the intended release path, but a + /// dropped-instead-of-destroyed event would strand its slot; enough of them exhaust the ring + /// and fail every later submission with `Busy` for the life of the instance. /// /// Only a terminal slot is reclaimed. A `PENDING` slot still belongs to the dispatch worker, /// which will latch it; freeing it here would hand a live slot to the next submission. Such a diff --git a/crates/virtio-accel-xdna/tests/hardware.rs b/crates/virtio-accel-xdna/tests/hardware.rs index fb0facf..2cbbf4c 100644 --- a/crates/virtio-accel-xdna/tests/hardware.rs +++ b/crates/virtio-accel-xdna/tests/hardware.rs @@ -29,8 +29,8 @@ use virtio_accel_tosa::{ use virtio_accel_tosa_build::{OperatorKind, OwnedGraph, OwnedOperator, OwnedTensor}; use virtio_accel_xdna::{ XDNA_PRECOMPILED_FORMAT, XDNA_TOSA_FP8_TARGET, XDNA_TOSA_INTEGER_TARGET, XDNA_TOSA_TARGET, - XdnaAccelerator, XdnaBuffer, XdnaContext, XdnaProgram, XdnaQueue, XdnaResourceCounts, - compile_artifact, + XdnaAccelerator, XdnaBuffer, XdnaContext, XdnaEvent, XdnaProgram, XdnaQueue, + XdnaResourceCounts, compile_artifact, }; #[cfg(feature = "test-control")] use virtio_accel_xdna::{XdnaTestConfig, XdnaTestFault}; @@ -425,39 +425,321 @@ fn malformed_and_unsupported_artifacts_leave_no_native_resources() { } #[test] -fn concurrent_submit_is_rejected_without_disturbing_the_accepted_job() { +fn ring_capacity_bounds_outstanding_events_and_reclaim_restores_it() { let Some(backend) = backend() else { return }; - let resources = PassthroughResources::create(&backend); - let bindings = resources.bindings(); - let event = backend - .submit( - &resources.queue, - &resources.program, - &bindings, - Timeout::Infinite, + let context = backend + .create_context(ContextDesc::default()) + .expect("context"); + let queue = backend + .create_queue(&context, QueueDesc::default()) + .expect("queue"); + let program = backend + .load_program( + &context, + ArtifactRef { + format: XDNA_PRECOMPILED_FORMAT, + target: TargetIdentity([0; 12]), + payload: &Slice(PASSTHROUGH), + resident_bytes: u64::MAX, + }, ) - .expect("first submit"); + .expect("load passthrough"); + let input_desc = BufferDesc::new( + PASSTHROUGH_BYTES as u64, + 4096, + MemoryDomain::Shared, + BufferUsage::TRANSFER_DESTINATION | BufferUsage::PROGRAM_INPUT, + ) + .expect("input descriptor"); + let output_desc = BufferDesc::new( + PASSTHROUGH_BYTES as u64, + 4096, + MemoryDomain::Shared, + BufferUsage::TRANSFER_SOURCE | BufferUsage::PROGRAM_OUTPUT, + ) + .expect("output descriptor"); + // Five distinct binding sets: the ring bound must come from the lane, not from binding the + // same buffers twice (the in-flight gates reject that separately). + const DEPTH: usize = 4; + let mut sets = Vec::new(); + for _ in 0..=DEPTH { + let (mut input, _) = backend + .allocate_buffer(&context, input_desc) + .expect("input") + .into_parts(); + let (unused, _) = backend + .allocate_buffer(&context, input_desc) + .expect("unused") + .into_parts(); + let (output, _) = backend + .allocate_buffer(&context, output_desc) + .expect("output") + .into_parts(); + let payload: Vec = (0..PASSTHROUGH_BYTES).map(|i| (i * 13 + 5) as u8).collect(); + backend + .write_buffer(&mut input, 0, &Slice(&payload)) + .expect("write input"); + sets.push((input, unused, output)); + } + let range = BufferRange::new(0, PASSTHROUGH_BYTES as u64).expect("binding range"); + fn bindings( + set: &(XdnaBuffer, XdnaBuffer, XdnaBuffer), + range: BufferRange, + ) -> [BindingRef<'_, XdnaBuffer>; 3] { + [ + BindingRef { + slot: 0, + buffer: &set.0, + range, + access: AccessMode::Read, + }, + BindingRef { + slot: 1, + buffer: &set.1, + range, + access: AccessMode::Read, + }, + BindingRef { + slot: 2, + buffer: &set.2, + range, + access: AccessMode::Write, + }, + ] + } + + // Fill the ring: DEPTH accepted submissions, polled to terminal but not destroyed, so every + // event slot stays claimed regardless of retirement timing. + let mut events = Vec::new(); + for set in &sets[..DEPTH] { + let event = backend + .submit(&queue, &program, &bindings(set, range), Timeout::Infinite) + .expect("submit within ring capacity"); + events.push(event); + } + for (index, event) in events.iter().enumerate() { + let state = poll_to_terminal(&backend, event, Duration::from_secs(10)) + .expect("accepted job did not complete"); + assert!( + matches!(state, EventState::Complete), + "job {index}: {state:?}" + ); + } + assert_eq!(backend.resource_counts().events, DEPTH as u64); + // Every slot is terminal-but-live: the next submission must reject with Busy and must not + // disturb the completed jobs. assert!(matches!( backend.submit( - &resources.queue, - &resources.program, - &bindings, - Timeout::Infinite, + &queue, + &program, + &bindings(&sets[DEPTH], range), + Timeout::Infinite ), Err(SubmitFailure::Rejected(BackendError::Busy)) )); - assert_eq!(backend.resource_counts().events, 1); - let state = poll_to_terminal(&backend, &event, Duration::from_secs(10)) - .expect("accepted job did not complete"); + // Reclaiming one slot restores capacity for exactly one more submission. + backend + .destroy_event(events.pop().expect("filled ring")) + .expect("destroy event"); + let refill = backend + .submit( + &queue, + &program, + &bindings(&sets[DEPTH], range), + Timeout::Infinite, + ) + .expect("submit after reclaim"); + let state = poll_to_terminal(&backend, &refill, Duration::from_secs(10)) + .expect("refill job did not complete"); assert!(matches!(state, EventState::Complete), "got {state:?}"); - backend.destroy_event(event).expect("destroy event"); - resources.release(&backend); + backend.destroy_event(refill).expect("destroy refill event"); + for event in events { + backend.destroy_event(event).expect("destroy event"); + } + for (input, unused, output) in sets { + backend.free_buffer(input).expect("free input"); + backend.free_buffer(unused).expect("free unused"); + backend.free_buffer(output).expect("free output"); + } + backend.unload_program(program).expect("unload"); + backend.destroy_queue(queue).expect("destroy queue"); + backend.destroy_context(context).expect("destroy context"); assert_eq!(backend.resource_counts(), XdnaResourceCounts::default()); } +/// Pipelined submissions must not mix jobs up: with the ring holding several in-flight +/// passthrough copies at once, every completion must carry exactly its own submission's payload. +/// Each round rotates distinct per-set payloads, so a cross-job binding or retirement-order bug +/// surfaces as a byte mismatch. #[test] -fn pending_releases_return_the_same_live_resources_for_retry() { +fn pipelined_submissions_complete_in_order_with_their_own_payloads() { let Some(backend) = backend() else { return }; + let context = backend + .create_context(ContextDesc::default()) + .expect("context"); + let queue = backend + .create_queue(&context, QueueDesc::default()) + .expect("queue"); + let program = backend + .load_program( + &context, + ArtifactRef { + format: XDNA_PRECOMPILED_FORMAT, + target: TargetIdentity([0; 12]), + payload: &Slice(PASSTHROUGH), + resident_bytes: u64::MAX, + }, + ) + .expect("load passthrough"); + let input_desc = BufferDesc::new( + PASSTHROUGH_BYTES as u64, + 4096, + MemoryDomain::Shared, + BufferUsage::TRANSFER_DESTINATION | BufferUsage::PROGRAM_INPUT, + ) + .expect("input descriptor"); + let output_desc = BufferDesc::new( + PASSTHROUGH_BYTES as u64, + 4096, + MemoryDomain::Shared, + BufferUsage::TRANSFER_SOURCE | BufferUsage::PROGRAM_OUTPUT, + ) + .expect("output descriptor"); + const SETS: usize = 4; + const ROUNDS: usize = 8; + let mut sets = Vec::new(); + for _ in 0..SETS { + let (input, _) = backend + .allocate_buffer(&context, input_desc) + .expect("input") + .into_parts(); + let (unused, _) = backend + .allocate_buffer(&context, input_desc) + .expect("unused") + .into_parts(); + let (output, _) = backend + .allocate_buffer(&context, output_desc) + .expect("output") + .into_parts(); + sets.push((input, unused, output)); + } + let range = BufferRange::new(0, PASSTHROUGH_BYTES as u64).expect("binding range"); + let payload_for = |round: usize, set: usize| -> Vec { + (0..PASSTHROUGH_BYTES) + .map(|i| (i.wrapping_mul(31) + round * 41 + set * 17 + 7) as u8) + .collect() + }; + + // Prefill: one in-flight job per set, then retire-verify-resubmit in submission order so up + // to `SETS` jobs overlap on the stream at every point. + let mut in_flight: std::collections::VecDeque<(usize, usize, XdnaEvent)> = + std::collections::VecDeque::new(); + for (set, entry) in sets.iter_mut().enumerate() { + backend + .write_buffer(&mut entry.0, 0, &Slice(&payload_for(0, set))) + .expect("write input"); + let bindings = [ + BindingRef { + slot: 0, + buffer: &entry.0, + range, + access: AccessMode::Read, + }, + BindingRef { + slot: 1, + buffer: &entry.1, + range, + access: AccessMode::Read, + }, + BindingRef { + slot: 2, + buffer: &entry.2, + range, + access: AccessMode::Write, + }, + ]; + let event = backend + .submit(&queue, &program, &bindings, Timeout::Infinite) + .expect("prefill submit"); + in_flight.push_back((0, set, event)); + } + let mut completed = 0usize; + while let Some((round, set, event)) = in_flight.pop_front() { + let state = poll_to_terminal(&backend, &event, Duration::from_secs(10)) + .expect("pipelined job did not complete"); + assert!( + matches!(state, EventState::Complete), + "round {round} set {set}: {state:?}" + ); + backend.destroy_event(event).expect("destroy event"); + let mut result = vec![0u8; PASSTHROUGH_BYTES]; + backend + .read_buffer(&sets[set].2, 0, &mut SliceMut(&mut result)) + .expect("read output"); + assert_eq!( + result, + payload_for(round, set), + "round {round} set {set}: output does not match its own submission's payload" + ); + completed += 1; + let next_round = round + 1; + if next_round < ROUNDS { + let entry = &mut sets[set]; + backend + .write_buffer(&mut entry.0, 0, &Slice(&payload_for(next_round, set))) + .expect("write input"); + let bindings = [ + BindingRef { + slot: 0, + buffer: &entry.0, + range, + access: AccessMode::Read, + }, + BindingRef { + slot: 1, + buffer: &entry.1, + range, + access: AccessMode::Read, + }, + BindingRef { + slot: 2, + buffer: &entry.2, + range, + access: AccessMode::Write, + }, + ]; + let event = backend + .submit(&queue, &program, &bindings, Timeout::Infinite) + .expect("pipelined resubmit"); + in_flight.push_back((next_round, set, event)); + } + } + assert_eq!(completed, SETS * ROUNDS); + + for (input, unused, output) in sets { + backend.free_buffer(input).expect("free input"); + backend.free_buffer(unused).expect("free unused"); + backend.free_buffer(output).expect("free output"); + } + backend.unload_program(program).expect("unload"); + backend.destroy_queue(queue).expect("destroy queue"); + backend.destroy_context(context).expect("destroy context"); + assert_eq!(backend.resource_counts(), XdnaResourceCounts::default()); +} + +#[cfg(feature = "test-control")] +#[test] +fn pending_releases_return_the_same_live_resources_for_retry() { + // A deterministic pending window: the worker holds this dispatch for 300ms, so the release + // rejections below never race real completion (which pipelined dispatch made fast enough to + // win occasionally). + let backend = XdnaAccelerator::new_for_testing(XdnaTestConfig { + watchdog_timeout: Duration::from_secs(2), + fault: XdnaTestFault::HoldDispatch { + hold: Duration::from_millis(300), + }, + }) + .expect("hold-controlled backend"); let PassthroughResources { context, queue, @@ -695,7 +977,7 @@ fn dropping_a_completed_event_reclaims_its_ring_slot() { }, ]; - // Two rounds through the single ring slot, releasing the first event by dropping it. + // Two rounds through one ring slot, releasing the first event by dropping it. for round in 0..2 { let event = match backend.submit(&queue, &program, &bindings, Timeout::Infinite) { Ok(event) => event, @@ -811,7 +1093,7 @@ fn precompiled_passthrough_runs_the_full_lifecycle() { Err(failure) => panic!("submit rejected: {failure:?}"), }; - // Poll to a terminal state (nonblocking poll; the worker bridges the blocking synchronize). + // Poll to a terminal state (nonblocking poll; the worker bridges the completion wait). let state = poll_to_terminal(&backend, &event, Duration::from_secs(10)) .expect("dispatch did not complete in 10s"); assert!( @@ -1509,6 +1791,215 @@ fn tosa_int32_to_int8_rescale_matches_the_shared_exact_oracle_on_the_npu() { backend.destroy_context(context).expect("destroy context"); } +/// Pipelined throughput of the exact INT8 MATMUL: four distinct buffer sets kept in flight, so +/// consecutive submissions overlap on the stream and the per-submission host cost amortizes. +/// Every warmup and every final in-flight completion is validated against the shared exact +/// oracle; the timed window itself is unvalidated (its buffers are rebound before the window +/// ends), which the printed line states. +#[test] +#[ignore = "manual native performance evidence"] +fn measures_pipelined_int8_matmul_throughput() { + const M: usize = 64; + const K: usize = 64; + const N: usize = 32; + const LEFT_ZP: i8 = -2; + const RIGHT_ZP: i8 = 3; + const SETS: usize = 4; + const WARMUP_ROUNDS: usize = 10; + const MEASURED: usize = 400; + let Some(backend) = backend() else { return }; + if !toolchain_present() { + assert!( + !hardware_required(), + "{REQUIRE_HARDWARE_ENV}=1 but VIRTIO_ACCEL_AMDXDNA_TOOLCHAIN is not configured" + ); + eprintln!("no XDNA toolchain configured; skipping INT8 pipeline benchmark"); + return; + } + let context = backend + .create_context(ContextDesc::default()) + .expect("context"); + let queue = backend + .create_queue(&context, QueueDesc::default()) + .expect("queue"); + let tosa = int8_matmul_tosa(M as i32, K as i32, N as i32, LEFT_ZP, RIGHT_ZP); + let program = backend + .load_program( + &context, + ArtifactRef { + format: ARTIFACT_FORMAT, + target: XDNA_TOSA_INTEGER_TARGET.to_identity(), + payload: &Slice(&tosa), + resident_bytes: u64::MAX, + }, + ) + .expect("load INT8 matmul"); + + let lhs_len = M * K; + let rhs_len = K * N; + let output_len = M * N * 4; + let input_desc = |bytes: usize| { + BufferDesc::new( + bytes as u64, + 4096, + MemoryDomain::Shared, + BufferUsage::TRANSFER_DESTINATION | BufferUsage::PROGRAM_INPUT, + ) + .unwrap() + }; + let mut sets = Vec::new(); + for set in 0..SETS { + let (mut lhs, _) = backend + .allocate_buffer(&context, input_desc(lhs_len)) + .expect("lhs") + .into_parts(); + let (mut rhs, _) = backend + .allocate_buffer(&context, input_desc(rhs_len)) + .expect("rhs") + .into_parts(); + let (output, _) = backend + .allocate_buffer( + &context, + BufferDesc::new( + output_len as u64, + 4096, + MemoryDomain::Shared, + BufferUsage::TRANSFER_SOURCE | BufferUsage::PROGRAM_OUTPUT, + ) + .unwrap(), + ) + .expect("output") + .into_parts(); + let lhs_bytes: Vec = (0..lhs_len) + .map(|i| (i * 37 + 11 + set * 3) as u8) + .collect(); + let rhs_bytes: Vec = (0..rhs_len) + .map(|i| (i * 53 + 197 + set * 7) as u8) + .collect(); + backend + .write_buffer(&mut lhs, 0, &Slice(&lhs_bytes)) + .expect("write lhs"); + backend + .write_buffer(&mut rhs, 0, &Slice(&rhs_bytes)) + .expect("write rhs"); + sets.push((lhs, rhs, output, lhs_bytes, rhs_bytes)); + } + let range = |bytes: usize| BufferRange::new(0, bytes as u64).unwrap(); + let submit_set = |set: &(XdnaBuffer, XdnaBuffer, XdnaBuffer, Vec, Vec)| { + backend + .submit( + &queue, + &program, + &[ + BindingRef { + slot: 0, + buffer: &set.0, + range: range(lhs_len), + access: AccessMode::Read, + }, + BindingRef { + slot: 1, + buffer: &set.1, + range: range(rhs_len), + access: AccessMode::Read, + }, + BindingRef { + slot: 2, + buffer: &set.2, + range: range(output_len), + access: AccessMode::Write, + }, + ], + Timeout::Infinite, + ) + .expect("pipelined submit") + }; + let verify_set = |set: &(XdnaBuffer, XdnaBuffer, XdnaBuffer, Vec, Vec)| { + let mut result = vec![0u8; output_len]; + backend + .read_buffer(&set.2, 0, &mut SliceMut(&mut result)) + .expect("read output"); + for row in 0..M { + for column in 0..N { + let left_row = &set.3[row * K..(row + 1) * K]; + let right_column: Vec = (0..K).map(|i| set.4[i * N + column]).collect(); + let expected = + dot_i8_i32(left_row, &right_column, LEFT_ZP, RIGHT_ZP, 0).expect("oracle"); + let actual = i32::from_le_bytes( + result[(row * N + column) * 4..][..4] + .try_into() + .expect("chunk"), + ); + assert_eq!(actual, expected, "C[{row},{column}] oracle mismatch"); + } + } + }; + + // Warmups: sequential rounds of the full pipeline, each completion oracle-validated. + for _ in 0..WARMUP_ROUNDS { + let mut events = std::collections::VecDeque::new(); + for set in &sets { + events.push_back(submit_set(set)); + } + for (index, event) in events.into_iter().enumerate() { + let state = poll_to_terminal(&backend, &event, Duration::from_secs(10)) + .expect("warmup completion"); + assert!(matches!(state, EventState::Complete), "warmup: {state:?}"); + backend.destroy_event(event).expect("destroy warmup event"); + verify_set(&sets[index]); + } + } + + // Timed window: keep the ring full; count completions. + let started = Instant::now(); + let mut in_flight: std::collections::VecDeque<(usize, _)> = std::collections::VecDeque::new(); + for (index, set) in sets.iter().enumerate() { + in_flight.push_back((index, submit_set(set))); + } + let mut completed = 0usize; + while completed < MEASURED { + let (index, event) = in_flight.pop_front().expect("in-flight event"); + let state = poll_to_terminal(&backend, &event, Duration::from_secs(10)) + .expect("pipelined completion"); + assert!(matches!(state, EventState::Complete), "measured: {state:?}"); + backend.destroy_event(event).expect("destroy event"); + completed += 1; + if completed + in_flight.len() < MEASURED { + in_flight.push_back((index, submit_set(&sets[index]))); + } + } + let elapsed = started.elapsed(); + assert!(in_flight.is_empty()); + + // The final in-flight generation is still in the output buffers: validate it. + for set in &sets { + verify_set(set); + } + + let per_inference = elapsed.as_secs_f64() / MEASURED as f64; + // Two operations per multiply-accumulate, matching `measures_exact_int8_matmul_latency`. + let ops = 2.0 * (M * K * N) as f64; + let gops = ops / per_inference / 1e9; + println!( + "XDNA pipelined exact INT8 MATMUL: shape=1x{M}x{K} . 1x{K}x{N}; \ + zero_points=[{LEFT_ZP},{RIGHT_ZP}]; ring_depth=4; sets={SETS}; \ + warmup_rounds={WARMUP_ROUNDS} (validated); measured={MEASURED} completions \ + (window unvalidated; final generation validated); \ + amortized={:.3}us/inference; effective={:.3} GOPS", + per_inference * 1e6, + gops, + ); + + for (lhs, rhs, output, _, _) in sets { + backend.free_buffer(lhs).expect("free lhs"); + backend.free_buffer(rhs).expect("free rhs"); + backend.free_buffer(output).expect("free output"); + } + backend.unload_program(program).expect("unload"); + backend.destroy_queue(queue).expect("destroy queue"); + backend.destroy_context(context).expect("destroy context"); +} + #[test] #[ignore = "manual native performance evidence"] fn measures_exact_int8_matmul_latency() { diff --git a/docs/performance.md b/docs/performance.md index 8900b62..2f0424f 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -213,6 +213,20 @@ per-submission overhead floor (the 1,024-element FP8 case measures 86 microsecon remaining latency is submission-path cost, not kernel cost. Issue #151 tracks the next steps (submission overlap, worker striping) without weakening exactness or direct binding. +The pipelined-throughput benchmark keeps four submissions in flight over four rotating buffer +sets (`measures_pipelined_int8_matmul_throughput`, same shape and oracle). On August 27, 2026 it +measured 73.3-74.9 microseconds amortized per inference across three 400-completion runs -- +statistically identical to the sequential submit-to-complete median (69.3-74.6 microseconds +across three runs of the latency benchmark on the same worker). A batched-flush variant (all +in-flight dispatches submitted under one `hrx_stream_flush`) measured 70.1 microseconds, also +identical. The conclusion this evidence supports: the per-submission floor is per-command +driver/firmware round-trip cost inside one hardware context, and neither deeper host-side +pipelining nor flush batching moves it. Raising effective throughput therefore requires more work +per dispatch (larger admitted envelopes, worker striping -- issue #151 steps 5-6) or parallel +hardware contexts (issue #121), not further submission-path restructuring. The ring depth of four +still pays for itself in semantics: submissions overlap with host-side polling and readback, and +completion waits no longer serialize against `allocate_buffer`. + ## Qualcomm Hexagon evidence status `virtio-accel-hexagon` includes an ignored release-mode measurement for fixed submission overhead: