//! Worker-local io_uring submission, completion, and deadlines. //! //! [`Waiters`] owns each request's descriptors, buffers, progress, and observer. //! The driver queues only IDs. Durability requests first acquire their open's //! permit. Eligible requests and partial completions join the ready queue, //! waiting for both SQ space and an available operation slot. //! Completion releases that slot even if the local result remains unconsumed //! in its waiter. Detached sync results leave through [`Deferred`]. //! //! # Submission and Cancellation //! //! Staging keeps every SQE-referenced resource alive until the operation CQE, //! including when submission returns a transient error. Cancellation CQEs only //! acknowledge the cancellation attempt. They cannot release those resources. //! //! Dropping a local or forwarded observer cancels reads and network operations. //! Queued requests can finish immediately, while in-flight requests need a cancellation //! SQE and must await their operation CQE. Writes and syncs continue through all //! follow-up SQEs after their observers are dropped. //! //! Cancellations stage first, followed by operations in FIFO order and wake-poll //! rearming. The operation limit excludes cancellation and wake SQEs. A full SQ //! needs a flush, but reaching the operation limit requires a CQE to free capacity. //! //! # Service and Parking //! //! Each service turn reaps completions, advances the timeout wheel, registers new //! deadlines, and stages queued work. Reaping first lets a terminal completion //! win a race with expiry. Partial progress after expiry cannot issue another SQE. //! Queued requests expire even while all operation slots are occupied. //! //! The ring uses SINGLE_ISSUER and DEFER_TASKRUN, requiring Linux 6.1 or newer. //! Unfinished requests need a GETEVENTS enter even when tasks keep the worker //! busy. An idle turn may defer that enter to [`Driver::park`]. Control SQEs carry //! no request resources and may remain until later I/O service or ring teardown. //! //! Parking installs the wake poll if needed, then arms the mailbox waker and //! checks for pending publication or a latched signal before blocking. A wake //! CQE without MORE requires rearm before the next blocking wait. See //! [`super::idle`] for how pool worker selection fits with this wait. //! //! # Cleanup //! //! Observer callbacks and request-resource destruction go through [`Deferred`], //! outside the worker borrow. Closure detaches local and forwarded observers and //! cancels eligible requests. The worker continues servicing until all requests //! retire, keeping the driver in place throughout the drain. use super::{ registration::Observation, request::{Request, RequestOutput}, runtime::{Deferred, RingConfig}, timeout::TimeoutWheel, waiter::{CompletionOutcome, Observer, UserData, WaiterId, Waiters}, waker::{WAKE_USER_DATA, Waker}, }; use crate::Error; use commonware_utils::{channel::oneshot, sync::OwnedAsyncMutexGuard}; use futures::{Stream, future::BoxFuture, stream::FuturesUnordered}; use io_uring::{ EnterFlags, IoUring, opcode::AsyncCancel, squeue::SubmissionQueue, types::{SubmitArgs, Timespec}, }; use std::{ collections::VecDeque, io, pin::Pin, sync::Arc, task::{Context, Poll, Waker as TaskWaker}, time::{Duration, Instant}, }; use tracing::warn; /// Shared acknowledgement token for cancellation SQEs. /// /// Like the wake token, this uses the slot index reserved by [`Waiters::insert`]. const CANCEL_USER_DATA: UserData = u32::MAX as UserData; /// Ring and request state accessed exclusively by the owning worker. pub struct Driver { /// Declared first so ring destruction precedes descriptor and buffer release. ring: IoUring, /// Kept separately to allow batching SQ access while updating request state. state: State, } /// Request state borrowed independently of the ring's SQ and CQ mappings. struct State { /// Queued requests, in-flight requests, and unconsumed local results. waiters: Waiters, /// Maximum number of outstanding operation SQEs. in_flight_limit: usize, /// FIFO of initial and follow-up operation SQEs, with lazily removed stale IDs. ready_queue: VecDeque, /// Accepted durability requests awaiting their open's permit, absent from the FIFO. acquiring: FuturesUnordered)>>, /// Permit grants wake ring service even after the ordinary mailbox closes. acquisition_waker: TaskWaker, /// New requests whose deadlines must be registered after advancing the wheel. pending_deadlines: VecDeque, /// Requests awaiting a cancellation SQE, skipped if their operation finishes first. pending_cancels: VecDeque, /// Deadlines for queued and in-flight requests, independent of sleepers. timeout_wheel: TimeoutWheel, /// Shared mailbox wake source retained through ring destruction. waker: Waker, /// Whether the multishot wake poll needs to be installed again. wake_rearm_needed: bool, } /// What [`Driver::enter`] asks of the kernel after submitting queued SQEs. #[derive(Clone, Copy)] enum Enter { /// Nothing more. Used to make room in a full SQ. Flush, /// Run deferred task work, which posts finished requests to the CQ, and /// return without waiting. TaskWork, /// Run deferred task work and sleep until a completion arrives. Wait { /// When to stop waiting, or `None` to wait without a bound. A /// deadline that already passed does not sleep. deadline: Option, }, } impl Driver { /// Create a ring on its permanent owner thread after configuration validation. /// /// Uses the CQ sizing and overflow policy described by [`RingConfig::size`]. pub fn new( cfg: &RingConfig, max_timeout: Duration, waker: Waker, now: Instant, ) -> io::Result { assert!(cfg.size > 0 && cfg.size.is_power_of_two() && cfg.size <= 32_768); TimeoutWheel::validate_layout(max_timeout, cfg.timeout_wheel_tick) .expect("validated timeout wheel configuration"); // Deferred task work runs during GETEVENTS, supplied by service or park. let ring = IoUring::builder() .setup_single_issuer() .setup_defer_taskrun() .build(cfg.size)?; let size = cfg.size as usize; Ok(Self { ring, state: State { waiters: Waiters::new(size), in_flight_limit: size, ready_queue: VecDeque::with_capacity(size), acquiring: FuturesUnordered::new(), acquisition_waker: Arc::new(waker.clone()).into(), pending_deadlines: VecDeque::with_capacity(size), pending_cancels: VecDeque::with_capacity(size), timeout_wheel: TimeoutWheel::new(max_timeout, cfg.timeout_wheel_tick, now), waker, wake_rearm_needed: true, }, }) } /// Accept an owned request without staging kernel work. /// /// Cached time can reject expired requests immediately. Service validates /// remaining deadlines after the wheel has advanced. pub fn admit( &mut self, request: Request, observer: Observer, now: Instant, deferred: &mut Deferred, ) -> WaiterId { let deadline = request.deadline(); let durability = request.durability().cloned(); // Transfer ownership before any wait for staging capacity or durability. let id = self.state.waiters.insert(request, observer); if let Some(deadline) = deadline { if deadline <= now { self.state.complete(id, Err(Error::Timeout), deferred); return id; } // A task poll may have taken long enough to leave the wheel behind. // Service refreshes it before assigning this deadline a tick. self.state.pending_deadlines.push_back(id); } // A durability request holds the open's permit through terminal accounting so a later // barrier observes any earlier failure. Contended requests wait in `acquiring` outside // the ready queue. They have no deadline and finish even if their caller is dropped. if let Some(durability) = durability { match durability.clone().try_lock_owned() { Ok(permit) => self.state.acquired(id, permit, deferred), Err(_) => { // The native driver owns this wait, so no cooperative budget // of a task polling the driver may delay it. Admission may // follow the last service turn before parking, so poll now // to register the foreign release's wake target. self.state.acquiring.push(Box::pin(async move { ( id, tokio::task::unconstrained(durability.lock_owned()).await, ) })); self.state.poll_acquisitions(deferred); } } } else { self.state.ready_queue.push_back(id); } id } /// Number of active logical requests, excluding retained completed outputs. pub const fn len(&self) -> usize { self.state.waiters.len() } /// Whether all logical requests retired, excluding retained completed outputs. pub const fn is_empty(&self) -> bool { self.state.waiters.is_empty() } /// Whether actionable staging or deadline registration must run before parking. pub fn has_pending_submissions(&self) -> bool { (!self.state.ready_queue.is_empty() && self.state.waiters.in_flight() < self.state.in_flight_limit) || !self.state.pending_deadlines.is_empty() || !self.state.pending_cancels.is_empty() } /// Earliest active deadline registered on the operation wheel. pub fn next_deadline(&self) -> Option { self.state.timeout_wheel.next_deadline() } /// Consume a retained result or check whether its waker needs refreshing. pub fn observe(&mut self, id: WaiterId, waker: &TaskWaker) -> Observation { self.state.waiters.observe(id, waker) } /// Install a waker cloned outside the worker borrow, returning the displaced one. pub fn set_waker(&mut self, id: WaiterId, waker: TaskWaker) -> Option { self.state.waiters.set_waker(id, waker) } /// Forward local observation while retaining all kernel resources locally. pub fn forward( &mut self, id: WaiterId, sender: oneshot::Sender>, deferred: &mut Deferred, ) { self.state.waiters.forward(id, sender, deferred); } /// Detach local or forwarded observation and request eligible cancellation. pub fn orphan(&mut self, id: WaiterId, deferred: &mut Deferred) { if self.state.waiters.orphan(id, deferred) { self.state.cancel(id, deferred); } } /// Clear local and forwarded observation before draining retained writes and syncs. pub fn close(&mut self, deferred: &mut Deferred) { for id in self.state.waiters.close(deferred) { self.state.cancel(id, deferred); } } /// Process completions, deadlines, and queued submissions without callbacks. /// /// Returns whether this turn deferred GETEVENTS for unfinished requests. The /// `defer_kernel_service` argument permits the following idle ring wait to /// supply GETEVENTS. It is ignored when callbacks or unfinished staging /// already require another busy turn. pub fn service( &mut self, now: Instant, defer_kernel_service: bool, deferred: &mut Deferred, ) -> io::Result { // Finish posted work before expiring requests, then register deadlines // against the refreshed wheel so an idle interval cannot shorten them. self.state.reap(&mut self.ring, deferred); self.state.advance_timeouts(now, deferred); self.state.register_deadlines(now, deferred); self.state.poll_acquisitions(deferred); self.state.compact_ready_queue(); while self.state.fill_submission_queue(&mut self.ring) { let before = self.ring.submission().len(); Self::enter(&mut self.ring, Enter::Flush)?; if self.ring.submission().len() >= before { // A transient enter may leave the SQ full. Preserve queued // identities and give the kernel a completion-service point // before another staging attempt. break; } } // Busy turns must run deferred kernel work. An idle turn can leave this // enter to park, unless callbacks or more submissions need attention. let service_kernel = !self.is_empty() && (!defer_kernel_service || !deferred.is_empty() || self.has_pending_submissions()); if service_kernel { Self::enter(&mut self.ring, Enter::TaskWork)?; } // New CQEs can complete requests or terminate the multishot wake poll. // Park rechecks rearm before its next blocking enter. self.state.reap(&mut self.ring, deferred); #[cfg(test)] tests::after_service(&self.ring, deferred)?; Ok(!service_kernel && !self.is_empty()) } /// Wait for ring activity or the deadline while retaining worker ownership. /// /// Returns `false` when rearm needs another service turn or the publication /// handshake rejects sleeping. A skipped wait does not satisfy a deferred /// GETEVENTS requirement, so the worker must service nonblocking before its /// next task-polling turn when kernel work remains. pub fn park(&mut self, processed_seq: u32, deadline: Option) -> io::Result { if self.has_pending_submissions() { return Ok(false); } // Rearm before publishing the intention to sleep. A transient flush can // leave the SQ full, in which case service must run before trying again. if self.state.wake_rearm_needed { if !self.state.waker.reinstall(&mut self.ring.submission()) { Self::enter(&mut self.ring, Enter::Flush)?; if !self.state.waker.reinstall(&mut self.ring.submission()) { return Ok(false); } } self.state.wake_rearm_needed = false; } // Keep the arm guard alive through the wait. A publication racing this // check either prevents sleeping or signals the installed wake poll. let arm = self.state.waker.arm(processed_seq); if !arm.still_idle() { return Ok(false); } Self::enter(&mut self.ring, Enter::Wait { deadline })?; Ok(true) } /// Submit queued SQEs, then do what `mode` asks of the kernel. /// /// The caller reaps the CQ afterwards. A timeout or a transient error /// (`EINTR`, `EAGAIN`, `EBUSY`) returns `Ok` because the next service /// turn retries, so `Ok` does not mean the kernel consumed the SQ or posted /// a completion. Other errors are permanent, and once the kernel has /// dropped a completion every enter fails with `EBADR`. fn enter(ring: &mut IoUring, mut mode: Enter) -> io::Result<()> { #[cfg(test)] if matches!(mode, Enter::Flush) && tests::stall_flush(ring) { return Ok(()); } // Read the clock once, right before the syscall. A wait for one // completion with a zero timeout arms an expired timer and sleeps until // it fires whenever no completion is ready, so a deadline that already // passed runs only the task work. let mut timeout = Duration::ZERO; if let Enter::Wait { deadline: Some(deadline), } = mode { timeout = deadline.saturating_duration_since(Instant::now()); if timeout.is_zero() { mode = Enter::TaskWork; } } let result = match mode { // The crate adds GETEVENTS when the CQ has overflowed, so the kernel // also moves its overflow backlog into the CQ. Enter::Flush => ring.submit(), // GETEVENTS with no minimum completion count runs a batch of // deferred task work and returns, as liburing's // `io_uring_submit_and_get_events` does. Newer kernels bound the // batch and leave the rest for a later enter. Enter::TaskWork => { let to_submit = u32::try_from(ring.submission().len()).expect("SQ length must fit u32"); // SAFETY: `to_submit` counts only entries already published to // the SQ, whose resources stay alive until their CQEs, and no // argument pointer is passed. unsafe { ring.submitter().enter::( to_submit, 0, EnterFlags::GETEVENTS.bits(), None, ) } } Enter::Wait { deadline: None } => ring.submit_and_wait(1), Enter::Wait { deadline: Some(_) } => { // EXT_ARG bounds the wait without reserving an SQE for a timeout. let ts = Timespec::new() .sec(timeout.as_secs()) .nsec(timeout.subsec_nanos()); let args = SubmitArgs::new().timespec(&ts); match ring.submitter().submit_with_args(1, &args) { // This timeout bounds the enter. Request expiry belongs to // the wheel. Err(err) if err.raw_os_error() == Some(libc::ETIME) => Ok(0), result => result, } } }; // A transient error falls through and returns `Ok`, so the caller // reaps before retrying. An immediate retry here could spin while the // kernel needs CQ space to make further progress. if let Err(err) = result && !matches!( err.raw_os_error(), Some(libc::EINTR | libc::EAGAIN | libc::EBUSY) ) { return Err(err); } // The kernel reports a dropped completion only once, and only from an // enter that submitted no SQEs and then waited with no completion // ready. An enter that submits discards the wait's result, and a flush, // a task-work enter or a wait that finds completions queued never // reaches the check. The ring's overflow counter grows only when the // kernel drops a CQE and never decreases, so check it after every // enter. Every later enter fails too, since the driver cannot tell // which request lost its completion and a drain would otherwise wait // for it forever. let dropped = ring.completion().overflow() != 0; #[cfg(test)] let dropped = dropped || tests::drop_completion(ring); if dropped { return Err(io::Error::from_raw_os_error(libc::EBADR)); } Ok(()) } } impl State { /// Make a granted request eligible only after checking the open's retained outcome. fn acquired( &mut self, id: WaiterId, permit: OwnedAsyncMutexGuard<()>, deferred: &mut Deferred, ) { if let Some(result) = self.waiters.acquire_durability(id, permit) { self.complete(id, result, deferred); } else { self.ready_queue.push_back(id); } } /// Advance only notified permit waits, using a wake path that survives worker closure. fn poll_acquisitions(&mut self, deferred: &mut Deferred) { while let Poll::Ready(Some((id, permit))) = Pin::new(&mut self.acquiring) .poll_next(&mut Context::from_waker(&self.acquisition_waker)) { self.acquired(id, permit, deferred); } } /// Store the terminal result and defer resource destruction and callbacks. fn complete(&mut self, id: WaiterId, result: Result<(), Error>, deferred: &mut Deferred) { // Finish splits the request into its result and deferred resources. // A previously cancelled request has already released its wheel tick. if let Some(tick) = self.waiters.finish(id, result, deferred) { self.timeout_wheel.remove(tick); } } /// Request cancellation once, finishing locally if no SQE is in flight. fn cancel(&mut self, id: WaiterId, deferred: &mut Deferred) { // Cancellation clears the waiter's tick. Save it before the transition // so the wheel loses its active count exactly once. let tick = self.waiters.target_tick(id); if !self.waiters.cancel(id) { return; } if let Some(tick) = tick { self.timeout_wheel.remove(tick); } if self.waiters.is_in_flight(id) { self.pending_cancels.push_back(id); } else { // No kernel access remains, including between partial completions. self.complete(id, Err(Error::Timeout), deferred); } } /// Advance the wheel even when no active operation deadline remains. fn advance_timeouts(&mut self, now: Instant, deferred: &mut Deferred) { let Some(expired) = self.timeout_wheel.advance(now) else { return; }; for id in expired { self.cancel(id, deferred); } } /// Register deadlines for new requests against this turn's refreshed wheel. fn register_deadlines(&mut self, now: Instant, deferred: &mut Deferred) { while let Some(id) = self.pending_deadlines.pop_front() { // Cancellation may have retired the request before registration. let Some(deadline) = self.waiters.deadline(id) else { continue; }; match self.timeout_wheel.checked_target_tick(deadline, now) { Ok(Some(tick)) => { // Remember the same tick in the waiter so completion or // cancellation can remove its active count from the wheel. self.timeout_wheel.schedule(id, tick); self.waiters.set_deadline(id, tick); } // Time spent waiting for service can exhaust the deadline. // Complete before staging, leaving the ready ID for lazy removal. Ok(None) => self.complete(id, Err(Error::Timeout), deferred), Err(message) => { // An unsupported deadline fails through the request result // path before the kernel can reference this request. let error = Error::Io(io::Error::new(io::ErrorKind::InvalidInput, message).into()); self.complete(id, Err(error), deferred); } } } } /// Remove stale queue IDs once they exceed both 64 and the live queued count. fn compact_ready_queue(&mut self) { // Requests awaiting a permit have no ready ID. Every other pending // request without an in-flight SQE has exactly one. Extra IDs are stale. let queued = self.waiters.len() - self.waiters.in_flight() - self.acquiring.len(); let stale = self .ready_queue .len() .checked_sub(queued) .expect("queued request missing from ready queue"); if stale <= 64 || stale <= queued { return; } // Queued cancellations and expirations leave IDs behind even when no // staging capacity opens. Compact in place without disturbing FIFO order. self.ready_queue.retain(|id| self.waiters.is_pending(*id)); } /// Stage operation SQEs in FIFO order, returning whether an SQ flush is needed. fn stage_ready_requests(&mut self, submission_queue: &mut SubmissionQueue<'_>) -> bool { while self.waiters.in_flight() < self.in_flight_limit { let Some(id) = self.ready_queue.front().copied() else { return false; }; // Cancelled and expired requests leave their IDs behind. Drop them // without consuming SQ capacity or reporting a flush. if !self.waiters.is_pending(id) { self.ready_queue.pop_front(); continue; } if submission_queue.is_full() { return true; } self.ready_queue.pop_front(); let sqe = self.waiters.stage(id); // SAFETY: The waiter retains every referenced descriptor and buffer // until its operation CQE. The SQ has capacity, and no callback can // retire the waiter between constructing and pushing this SQE. unsafe { submission_queue .push(&sqe) .expect("checked operation SQ capacity"); } } // Only a completion can release this limit. A flush cannot, and // cancellation and mailbox wake SQEs must still be allowed to stage. false } /// Stage cancellation SQEs, returning whether an SQ flush is needed. fn stage_cancellations(&mut self, submission_queue: &mut SubmissionQueue<'_>) -> bool { while !submission_queue.is_full() { let Some(id) = self.pending_cancels.pop_front() else { return false; }; // The operation may have finished before its cancellation staged. if !self.waiters.is_in_flight(id) { continue; } let cancel = AsyncCancel::new(id.user_data()) .build() .user_data(CANCEL_USER_DATA); // SAFETY: AsyncCancel carries only the stable user_data identity. // Its target's waiter retains every kernel-visible owner, and the // loop checked SQ capacity before taking this queued identity. unsafe { submission_queue .push(&cancel) .expect("checked cancellation SQ capacity"); } } !self.pending_cancels.is_empty() } /// Fill available SQ slots and report whether more work requires a flush. fn fill_submission_queue(&mut self, ring: &mut IoUring) -> bool { // One SQ view batches tail publication until it is dropped on return. let mut submission_queue = ring.submission(); // Cancellation must progress even when all operation slots are occupied. if self.stage_cancellations(&mut submission_queue) { return true; } if self.stage_ready_requests(&mut submission_queue) { return true; } // Reaching the operation limit still leaves wake polling eligible. // If the SQ itself is full, retain the rearm flag across the flush. if self.wake_rearm_needed { if !self.waker.reinstall(&mut submission_queue) { return true; } self.wake_rearm_needed = false; } false } /// Reap every posted CQE without invoking observer callbacks. fn reap(&mut self, ring: &mut IoUring, deferred: &mut Deferred) { // Dropping this CQ view returns the consumed slots to the kernel. // Observer callbacks remain deferred until the worker borrow ends. for cqe in ring.completion() { self.handle_cqe(cqe.user_data(), cqe.result(), cqe.flags(), deferred); } } /// Apply a CQE without invoking observer callbacks. fn handle_cqe( &mut self, user_data: UserData, result: i32, flags: u32, deferred: &mut Deferred, ) { // The reserved mailbox token is outside the waiter's ID space. if user_data == WAKE_USER_DATA { assert!( result >= 0, "wake poll CQE failed: requires Linux 6.1+ multishot polling" ); // Clear eventfd readiness. The worker drains the actual messages // after this CQ batch and the local borrow have ended. self.waker.acknowledge(); if !io_uring::cqueue::more(flags) { // The kernel removed the multishot poll. Reinstall it before // the next blocking wait, even if this was service's final reap. self.wake_rearm_needed = true; } return; } // Only the target operation's CQE can retire its resource owners. if user_data == CANCEL_USER_DATA { if result == 0 { // Cancellation successful. } else if result == -libc::EALREADY { // Cancellation is no longer possible at this stage. The target // operation CQE should follow shortly. } else if result == -libc::ENOENT { // The target may already have completed and vacated its slot. } else if result == -libc::EINVAL { panic!("async cancel SQE rejected by kernel: EINVAL"); } else { warn!(result, "unexpected async cancel CQE result"); } return; } // Waiters releases an operation's in-flight count before returning its outcome. match self.waiters.on_completion(user_data, result) { // Partial progress rejoins the tail behind already queued work. CompletionOutcome::Requeue(id) => self.ready_queue.push_back(id), CompletionOutcome::Complete(id, result) => self.complete(id, result, deferred), } } } #[cfg(test)] pub mod tests { use super::*; use crate::{ IoBuf, IoBufMut, IoBufs, iouring::{ mailbox::Mailbox, request::{ Cache, DIRECT_ALIGN, IOVEC_BATCH_SIZE, ReadAtRequest, RecvRequest, RequestOutput, SyncRequest, WriteAtRequest, WriteAtState, }, waker::tests::wait_until_eventfd_armed, }, storage::{hold::Hold, iouring::Shared}, }; use commonware_utils::channel::oneshot; use std::{ cell::Cell, fs::{self, OpenOptions}, io::Write, os::{ fd::{AsRawFd, RawFd}, unix::{fs::OpenOptionsExt as _, net::UnixStream}, }, panic::{AssertUnwindSafe, catch_unwind}, path::PathBuf, sync::Arc, task::Wake, thread, }; /// Bound of a timed wait that times out. It leaves the deadline in the /// future when `Driver::enter` samples the clock on a loaded runner, so /// the wait takes the timed path rather than the elapsed one. const TIMED_WAIT: Duration = Duration::from_millis(10); /// Kernel outcomes that cannot be scheduled reliably with a real ring. #[derive(Clone, Copy, PartialEq, Eq)] enum Fault { /// A transient flush leaves every SQ entry untouched. StallFlush, /// Service fails with retired resources awaiting deferred cleanup. AfterCompletion, /// The kernel drops a CQE because it cannot allocate an overflow entry. DropCompletion, } thread_local! { /// One failure scoped to the calling thread and the selected ring. static FAULT: Cell> = const { Cell::new(None) }; } /// Clear an unused injection when its test scope exits or unwinds. pub struct FaultGuard; impl Drop for FaultGuard { fn drop(&mut self) { FAULT.set(None); } } /// Install one failure without changing the driver's production state. fn inject(driver: &Driver, fault: Fault) -> FaultGuard { assert!(FAULT.get().is_none(), "driver fault already installed"); FAULT.set(Some((driver.ring.as_raw_fd(), fault))); FaultGuard } /// Fail service after completion ownership has moved to deferred cleanup. pub fn fail_after_completion(driver: &Driver) -> FaultGuard { inject(driver, Fault::AfterCompletion) } /// Consume the selected fault once, leaving later cleanup turns unaffected. fn take_fault(ring: &IoUring, fault: Fault) -> bool { if FAULT.get() != Some((ring.as_raw_fd(), fault)) { return false; } FAULT.set(None); true } /// Model a transient flush before the kernel consumes any SQ entries. pub fn stall_flush(ring: &IoUring) -> bool { take_fault(ring, Fault::StallFlush) } /// Model a CQE the kernel dropped, which it counts in the ring's overflow /// counter. pub fn drop_completion(ring: &IoUring) -> bool { take_fault(ring, Fault::DropCompletion) } /// Fail service when retired resources are awaiting deferred cleanup. pub fn after_service(ring: &IoUring, deferred: &Deferred) -> io::Result<()> { if !deferred.resources.is_empty() && take_fault(ring, Fault::AfterCompletion) { return Err(io::Error::other( "injected service failure after completion", )); } Ok(()) } /// Destination of a result collected by the harness. enum TestObserver { /// Local result identified by a test-assigned tag. Local(u64), /// Detached sync awaiting explicit publication by the test. DetachedSync(oneshot::Sender>), /// Discarded result from an orphaned request. Orphaned, } /// Completed request and the observer that would receive its output. struct Completed { /// Destination retained for assertions or detached publication. observer: TestObserver, /// Request result removed from its waiter or deferred callbacks. output: RequestOutput, } /// Distinct waker allocation used to identify a local completion. struct Notify; // Each observer needs a distinct identity for matching its deferred wake. #[allow(clippy::manual_noop_waker)] impl Wake for Notify { fn wake(self: Arc) {} } /// Collect results while driving a real ring or injecting operation CQEs. struct Harness { /// Ring and request state under test. driver: Driver, /// Callbacks and resources detached by driver transitions. deferred: Deferred, /// Local observers matched by waker identity, with test-assigned tags. tracked: Vec<(WaiterId, u64, TaskWaker)>, /// Collected results retained for assertions. completed: Vec, /// Initial wheel time, also used as the origin for simulated deadlines. start: Instant, } impl Harness { /// Create a real ring whose SQ size also bounds in-flight operations. fn new(size: u32) -> Self { let start = Instant::now(); Self { driver: Driver::new( &RingConfig { size, timeout_wheel_tick: Duration::from_millis(5), }, Duration::from_secs(60), Waker::new().unwrap(), start, ) .unwrap(), deferred: Deferred::default(), tracked: Vec::new(), completed: Vec::new(), start, } } /// Queue a local request whose result `collect` will consume automatically. fn admit(&mut self, request: Request, tag: u64) -> WaiterId { let waker = TaskWaker::from(Arc::new(Notify)); let id = self.driver.admit( request, Observer::Local(Some(waker.clone())), self.start, &mut self.deferred, ); self.tracked.push((id, tag, waker)); id } /// Stage a request without submitting its SQE, for simulated completions. fn stage(&mut self, request: Request, tick: Option, tag: u64) -> WaiterId { let waker = TaskWaker::from(Arc::new(Notify)); let id = self .driver .state .waiters .insert(request, Observer::Local(Some(waker.clone()))); if let Some(tick) = tick { self.driver.state.timeout_wheel.schedule(id, tick); self.driver.state.waiters.set_deadline(id, tick); } // Keep the fixture out of the ready queue and the kernel. Only // simulated_completion may release this operation's in-flight count. self.driver.state.waiters.stage(id); self.tracked.push((id, tag, waker)); id } /// Collect deferred outputs and consume local results in wake order. fn collect(&mut self) { for output in self.deferred.outputs.drain(..) { self.completed.push(Completed { observer: TestObserver::Orphaned, output, }); } for waker in self.deferred.wakes.drain(..) { // Match by observer identity to preserve completion order even // when earlier requests have already vacated and reused slots. let index = self .tracked .iter() .position(|(_, _, current)| current.will_wake(&waker)) .unwrap(); let (id, tag, _) = self.tracked.swap_remove(index); let Observation::Ready(output) = self.driver.observe(id, &waker) else { panic!("completion wake without retained output"); }; self.completed.push(Completed { observer: TestObserver::Local(tag), output, }); waker.wake(); } // Orphaned observers leave a waker to discard without consuming a result. for waker in self.deferred.drops.drain(..) { if let Some(index) = self .tracked .iter() .position(|(_, _, current)| current.will_wake(&waker)) { self.tracked.swap_remove(index); } } self.deferred.resources.clear(); // Let tests delay detached publication until after the driver drains. for (sender, output) in self.deferred.completions.drain(..) { self.completed.push(Completed { observer: TestObserver::DetachedSync(sender), output: RequestOutput::Sync(output), }); } } /// Drop a local or forwarded observer and collect any immediately discarded output. fn orphan(&mut self, id: WaiterId) { self.driver.orphan(id, &mut self.deferred); self.collect(); } /// Service at a chosen time and collect results. fn service_at(&mut self, now: Instant, defer: bool) { self.driver.service(now, defer, &mut self.deferred).unwrap(); self.collect(); } /// Service using wall-clock time and run deferred kernel work immediately. fn service(&mut self) { self.service_at(Instant::now(), false); } /// Require a real wake CQE before any further submission can mask its absence. fn reap_wake(&mut self) { let cqe = self .driver .ring .completion() .next() .expect("missing wake CQE"); assert_eq!(cqe.user_data(), WAKE_USER_DATA); self.driver.state.handle_cqe( cqe.user_data(), cqe.result(), cqe.flags(), &mut self.deferred, ); } /// Drive until `count` results have been collected, failing if progress stalls. fn until(&mut self, count: usize) { let limit = Instant::now() + Duration::from_secs(10); while self.completed.len() < count { assert!(Instant::now() < limit, "driver completion stalled"); self.service(); thread::yield_now(); } } /// Notify observers of closure without trying to consume retained results. fn close(&mut self) { self.driver.close(&mut self.deferred); for waker in self.deferred.wakes.drain(..) { waker.wake(); } self.tracked.clear(); self.collect(); } /// Close local and forwarded observation and wait for request retirement. fn drain(&mut self) { self.close(); let limit = Instant::now() + Duration::from_secs(10); while !self.driver.is_empty() || self.driver.has_pending_submissions() { assert!(Instant::now() < limit, "driver retirement stalled"); self.service(); thread::yield_now(); } } /// Inject progress only for SQEs deliberately kept out of the real ring. /// /// This models the completion status and byte count without writing data. fn simulated_completion(&mut self, id: WaiterId, result: i32) { match self .driver .state .waiters .on_completion(id.user_data(), result) { CompletionOutcome::Complete(id, result) => { self.driver.state.complete(id, result, &mut self.deferred) } CompletionOutcome::Requeue(id) => self.driver.state.ready_queue.push_back(id), } self.collect(); } } /// Build an exact receive with initialized storage for payload assertions. fn recv(fd: UnixStream, len: usize, deadline: Option) -> Request { Request::Recv(RecvRequest { fd: Arc::new(fd.into()), buf: IoBufMut::zeroed(len), offset: 0, len, exact: true, deadline, }) } /// Borrow the initialized prefix reported by a successful receive. fn received(output: &Completed) -> &[u8] { match &output.output { RequestOutput::Recv(Ok((buf, len))) => &buf.as_ref()[..*len], _ => panic!("expected successful recv"), } } /// Create an isolated open for tests that control durability completion. fn make_file(name: &str) -> (PathBuf, Arc) { let directory = std::env::temp_dir().join(format!("commonware_driver_{name}_{}", std::process::id())); let hold = Hold::acquire(&directory).unwrap(); let file = OpenOptions::new() .read(true) .write(true) .create(true) .truncate(true) .open(directory.join("blob")) .unwrap(); (directory, Shared::detached(file, hold)) } #[test] fn test_unconsumed_results_leave_staging_and_deadline_tracking() { let mut harness = Harness::new(1); // Occupy the only operation slot so stale IDs cannot drain by staging. let (active, mut peer) = UnixStream::pair().unwrap(); harness.admit(recv(active, 1, None), 0); harness.service(); assert_eq!(harness.driver.state.waiters.in_flight(), 1); // Keep completed results in their slots while the sole operation slot // stays occupied. Their queue IDs must still count as stale. let mut results = Vec::new(); let deadline = harness.start + Duration::from_nanos(1); for _ in 0..65 { let (socket, _peer) = UnixStream::pair().unwrap(); results.push(harness.driver.admit( recv(socket, 1, Some(deadline)), Observer::Local(None), harness.start, &mut harness.deferred, )); } harness.service(); assert_eq!(harness.driver.len(), 1); assert!(harness.driver.state.ready_queue.is_empty()); assert!(harness.driver.state.pending_deadlines.is_empty()); assert!(harness.driver.next_deadline().is_none()); assert!(!harness.driver.has_pending_submissions()); // Below the compaction threshold, staging itself must skip the ID // even though its generation still matches a retained result. let (socket, _peer) = UnixStream::pair().unwrap(); results.push(harness.driver.admit( recv(socket, 1, Some(deadline)), Observer::Local(None), harness.start, &mut harness.deferred, )); harness.service(); assert_eq!(harness.driver.state.ready_queue.len(), 1); // Releasing capacity lets staging discard the one remaining stale ID. peer.write_all(b"x").unwrap(); harness.until(1); harness.service(); assert!(harness.driver.is_empty()); assert!(!harness.driver.has_pending_submissions()); // Neither compaction nor skipped staging may consume the results. for id in results { assert!(matches!( harness.driver.observe(id, TaskWaker::noop()), Observation::Ready(RequestOutput::Recv(Err((_, Error::Timeout)))) )); } harness.drain(); } #[test] fn test_cancelled_backlog_compacts_at_staging_limit() { let mut harness = Harness::new(1); let (active, mut active_peer) = UnixStream::pair().unwrap(); harness.admit(recv(active, 1, None), 0); harness.service(); // Cancellation must bound stale storage even while a live request at // the front prevents normal FIFO removal. let mut peak_queued = 0; for tag in 1..=2 { let (queued, mut peer) = UnixStream::pair().unwrap(); peer.write_all(b"x").unwrap(); harness.admit(recv(queued, 1, None), tag); // Stale IDs sit behind live requests. The second live request also // reuses a slot still named by earlier cancelled queue entries. for _ in 0..256 { let (socket, _peer) = UnixStream::pair().unwrap(); let id = harness.admit(recv(socket, 1, None), 3); harness.orphan(id); harness.service(); harness.completed.clear(); peak_queued = peak_queued.max(harness.driver.state.ready_queue.len()); } } assert_eq!(harness.driver.len(), 3); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(!harness.driver.has_pending_submissions()); // Compaction must preserve both live identities and their FIFO order. active_peer.write_all(b"x").unwrap(); harness.until(3); for (tag, completed) in harness.completed.iter().enumerate() { assert!(matches!(completed.observer, TestObserver::Local(id) if id == tag as u64)); assert_eq!(received(completed), b"x"); } assert!(peak_queued <= 128, "retained {peak_queued} queue entries"); harness.drain(); } #[test] fn test_expired_backlog_compacts_at_staging_limit() { let mut harness = Harness::new(1); let (active, _active_peer) = UnixStream::pair().unwrap(); harness.admit(recv(active, 1, None), 0); harness.service(); let (queued, _queued_peer) = UnixStream::pair().unwrap(); harness.admit(recv(queued, 1, None), 1); let mut now = harness.start; let mut peak_queued = 0; for register_first in [false, true] { for _ in 0..256 { let deadline = now + Duration::from_millis(10); let (socket, _peer) = UnixStream::pair().unwrap(); let id = harness.admit(recv(socket, 1, Some(deadline)), 2); // Exercise expiry during initial registration and through the // timeout wheel, while the active request prevents staging. if register_first { harness.service_at(now, false); } now = deadline; harness.service_at(now, false); assert!(!harness.driver.state.waiters.is_pending(id)); assert_eq!(harness.completed.len(), 1); assert!(matches!( harness.completed[0].output, RequestOutput::Recv(Err((_, Error::Timeout))) )); harness.completed.clear(); peak_queued = peak_queued.max(harness.driver.state.ready_queue.len()); } } assert_eq!(harness.driver.len(), 2); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(!harness.driver.has_pending_submissions()); assert!(peak_queued <= 128, "retained {peak_queued} queue entries"); harness.drain(); } #[test] fn test_staging_limit_parks_and_preserves_fifo_across_growth() { let mut harness = Harness::new(1); let (first, mut first_peer) = UnixStream::pair().unwrap(); let first = harness.admit(recv(first, 1, None), 0); harness.service(); assert!(harness.driver.state.waiters.is_in_flight(first)); // Grow the waiter slab while the kernel still holds the first buffer. for (tag, bytes) in [(1, b"b"), (2, b"c")] { let (socket, mut peer) = UnixStream::pair().unwrap(); peer.write_all(bytes).unwrap(); harness.admit(recv(socket, 1, None), tag); } harness.service(); assert_eq!(harness.driver.len(), 3); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(!harness.driver.has_pending_submissions()); // The backlog waits for a CQE. It must not prevent entering the idle wait. assert!(harness.driver.park(0, Some(Instant::now())).unwrap()); first_peer.write_all(b"a").unwrap(); harness.until(3); for (tag, expected) in [b"a", b"b", b"c"].into_iter().enumerate() { let completed = &harness.completed[tag]; assert!(matches!(completed.observer, TestObserver::Local(id) if id == tag as u64)); assert_eq!(received(completed), expected); } assert_eq!(harness.driver.state.waiters.in_flight(), 0); harness.drain(); } #[test] fn test_queued_expiry_and_cancellation_progress_at_staging_limit() { let mut harness = Harness::new(1); let (first, _first_peer) = UnixStream::pair().unwrap(); let first = harness.admit(recv(first, 1, None), 0); harness.service(); let (timed, _timed_peer) = UnixStream::pair().unwrap(); let timed = harness.admit( recv(timed, 1, Some(harness.start + Duration::from_secs(1))), 1, ); let (cancelled, _cancelled_peer) = UnixStream::pair().unwrap(); let cancelled = harness.admit(recv(cancelled, 1, None), 2); harness.orphan(cancelled); // Neither queued request needs an SQE to retire while the first is blocked. harness.service_at(harness.start + Duration::from_secs(2), false); assert!(!harness.driver.state.waiters.is_pending(cancelled)); assert!(!harness.driver.state.waiters.is_pending(timed)); assert_eq!(harness.completed.len(), 2); assert!(matches!( harness.completed[0].observer, TestObserver::Orphaned )); assert!(matches!( harness.completed[1].output, RequestOutput::Recv(Err((_, Error::Timeout))) )); assert!(harness.driver.state.waiters.is_in_flight(first)); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(!harness.driver.has_pending_submissions()); // Cancelling the active request must still stage its cancellation at the limit. harness.drain(); assert_eq!(harness.driver.state.waiters.in_flight(), 0); } #[test] fn test_unconsumed_result_releases_staging_capacity() { let mut harness = Harness::new(1); let (first, mut first_peer) = UnixStream::pair().unwrap(); first_peer.write_all(b"a").unwrap(); // Omit the observer waker so the harness leaves this result in Waiters. let first = harness.driver.admit( recv(first, 1, None), Observer::Local(None), harness.start, &mut harness.deferred, ); let (second, mut second_peer) = UnixStream::pair().unwrap(); second_peer.write_all(b"b").unwrap(); harness.admit(recv(second, 1, None), 1); // The second request needs the sole operation slot released by the first. harness.until(1); assert!(harness.driver.is_empty()); assert_eq!(harness.driver.state.waiters.in_flight(), 0); assert_eq!(received(&harness.completed[0]), b"b"); // Only now consume the first result, after the second has finished. assert!(matches!( harness.driver.observe(first, TaskWaker::noop()), Observation::Ready(RequestOutput::Recv(Ok((_, 1)))) )); harness.drain(); } #[test] fn test_expired_registration_never_stages() { let mut harness = Harness::new(1); let (left, _right) = UnixStream::pair().unwrap(); let deadline = harness.start + Duration::from_nanos(1); harness.admit(recv(left, 8, Some(deadline)), 0); // A deadline equal to this turn's time has already expired. harness.service_at(deadline, true); assert!(matches!( harness.completed[0].output, RequestOutput::Recv(Err((_, Error::Timeout))) )); assert!(harness.driver.is_empty()); assert!(harness.driver.next_deadline().is_none()); // Only the mailbox poll reached the SQ. The expired receive did not. assert_eq!(harness.driver.ring.submission().len(), 1); harness.drain(); } #[test] fn test_staging_discards_stale_ids_with_full_sq() { for with_pending in [false, true] { let mut harness = Harness::new(1); // The mailbox poll fills the SQ without using an operation slot. harness.service(); assert!(harness.driver.ring.submission().is_full()); let (socket, _peer) = UnixStream::pair().unwrap(); let deadline = harness.start + Duration::from_secs(10); let stale = harness.admit(recv(socket, 1, Some(deadline)), 0); harness.orphan(stale); let pending = with_pending.then(|| { let (socket, mut peer) = UnixStream::pair().unwrap(); peer.write_all(b"x").unwrap(); harness.admit(recv(socket, 1, None), 1) }); if let Some(pending) = pending { assert_eq!(stale.0.index, pending.0.index); assert_ne!(stale, pending); } let flush = harness .driver .state .stage_ready_requests(&mut harness.driver.ring.submission()); // Only a live request justifies a flush. A stale ID must disappear // even if there is no SQ space and its slot has already been reused. assert_eq!(flush, with_pending); assert_eq!( harness.driver.state.ready_queue.len(), usize::from(with_pending) ); assert_eq!(harness.driver.state.ready_queue.front().copied(), pending); assert_eq!(harness.driver.state.waiters.in_flight(), 0); assert!(harness.driver.ring.submission().is_full()); if with_pending { harness.until(2); assert_eq!(received(&harness.completed[1]), b"x"); } harness.drain(); // The old deadline registration must also be skipped after reuse. assert!(harness.driver.next_deadline().is_none()); assert!(harness.driver.state.pending_deadlines.is_empty()); } } #[test] fn test_advance_timeouts_ignores_stale_entry_after_slot_reuse() { let mut harness = Harness::new(1); let early = harness.start + Duration::from_millis(5); let late = harness.start + Duration::from_millis(15); let (old, _old_peer) = UnixStream::pair().unwrap(); let old = harness.admit(recv(old, 1, Some(early)), 0); let (survivor, _survivor_peer) = UnixStream::pair().unwrap(); let survivor = harness.admit(recv(survivor, 1, Some(early)), 1); harness .driver .state .register_deadlines(harness.start, &mut harness.deferred); // Keep another live deadline in the old bucket. Otherwise the wheel // can skip it entirely, without passing its stale record to the driver. harness.orphan(old); harness.completed.clear(); let (current, _current_peer) = UnixStream::pair().unwrap(); let current = harness.admit(recv(current, 1, Some(late)), 2); assert_eq!(old.0.index, current.0.index); assert_ne!(old, current); harness .driver .state .register_deadlines(harness.start, &mut harness.deferred); // The shared bucket expires, but its stale ID must not cancel the // replacement or remove the replacement's later deadline. harness .driver .state .advance_timeouts(early, &mut harness.deferred); harness.collect(); assert!(!harness.driver.state.waiters.is_pending(survivor)); assert!(harness.driver.state.waiters.is_pending(current)); assert_eq!(harness.completed.len(), 1); assert_eq!(harness.driver.next_deadline(), Some(late)); // The replacement must still expire at its own deadline. harness .driver .state .advance_timeouts(late, &mut harness.deferred); harness.collect(); assert!(harness.driver.is_empty()); assert!(harness.driver.next_deadline().is_none()); assert_eq!(harness.completed.len(), 2); for (tag, completed) in (1..=2).zip(&harness.completed) { assert!(matches!(completed.observer, TestObserver::Local(id) if id == tag)); assert!(matches!( completed.output, RequestOutput::Recv(Err((_, Error::Timeout))) )); } harness.drain(); } #[test] fn test_deadline_registration_uses_refreshed_wheel() { let mut harness = Harness::new(1); let now = harness.start + Duration::from_secs(3600) + Duration::from_millis(1); let deadline = now + Duration::from_secs(60); let (left, _right) = UnixStream::pair().unwrap(); let id = harness.admit(recv(left, 1, Some(deadline)), 0); // The wheel is an hour behind, but the deadline fits its horizon when // measured from this service turn. Refresh before validating it. harness.service_at(now, true); assert!(harness.completed.is_empty()); let scheduled = harness.driver.next_deadline().unwrap(); assert!(scheduled >= deadline); assert!(scheduled < deadline + Duration::from_millis(5)); harness.orphan(id); harness.drain(); } #[test] fn test_unsupported_deadline_is_rejected_without_submission() { let mut harness = Harness::new(1); let (left, _right) = UnixStream::pair().unwrap(); // This future deadline exceeds the harness's 60-second horizon. harness.admit( recv(left, 1, Some(harness.start + Duration::from_secs(61))), 0, ); harness.service_at(harness.start, true); assert!(matches!( harness.completed[0].output, RequestOutput::Recv(Err((_, Error::Io(_)))) )); assert!(harness.driver.is_empty()); assert_eq!(harness.driver.ring.submission().len(), 1); harness.drain(); } #[test] fn test_completion_wins_cancellation() { let mut harness = Harness::new(1); let (left, _right) = UnixStream::pair().unwrap(); let id = harness.stage(recv(left, 5, None), None, 0); harness.driver.state.cancel(id, &mut harness.deferred); // A cancellation request cannot undo a successful terminal CQE. harness.simulated_completion(id, 5); assert_eq!(received(&harness.completed[0]).len(), 5); harness.drain(); } #[test] fn test_cancel_acknowledgements_leave_requests_and_results_untouched() { let mut harness = Harness::new(1); let (left, _right) = UnixStream::pair().unwrap(); let id = harness.stage(recv(left, 5, None), None, 0); harness.driver.state.cancel(id, &mut harness.deferred); // Acknowledgements cannot release the operation's kernel resources. for result in [0, -libc::EALREADY, -libc::ENOENT, -libc::EPERM] { harness .driver .state .handle_cqe(CANCEL_USER_DATA, result, 0, &mut harness.deferred); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(harness.driver.state.waiters.is_pending(id)); } // Complete without collecting the result, leaving it in its waiter slot. let CompletionOutcome::Complete(current, result) = harness .driver .state .waiters .on_completion(id.user_data(), 5) else { panic!("expected terminal completion"); }; assert_eq!(current, id); harness .driver .state .complete(id, result, &mut harness.deferred); for result in [0, -libc::ENOENT] { harness .driver .state .handle_cqe(CANCEL_USER_DATA, result, 0, &mut harness.deferred); assert_eq!(harness.driver.state.waiters.in_flight(), 0); } harness.collect(); assert_eq!(harness.completed.len(), 1); assert_eq!(received(&harness.completed[0]).len(), 5); harness.drain(); } #[test] fn test_cancel_acknowledgement_invariants() { // Both control tokens use the index that Waiters refuses to allocate. assert_ne!(CANCEL_USER_DATA, WAKE_USER_DATA); for token in [CANCEL_USER_DATA, WAKE_USER_DATA] { assert_eq!(token as u32, u32::MAX); } let mut harness = Harness::new(1); // EINVAL rejects the cancellation SQE itself, regardless of slot reuse. assert!( catch_unwind(AssertUnwindSafe(|| { harness.driver.state.handle_cqe( CANCEL_USER_DATA, -libc::EINVAL, 0, &mut harness.deferred, ); })) .is_err() ); } #[test] fn test_request_retirement_releases_hold_before_cancel_acknowledgement() { for consume_cancel in [false, true] { for reuse in [false, true] { let mut harness = Harness::new(1); let directory = std::env::temp_dir().join(format!( "commonware_driver_retirement_{}_{}_{}", std::process::id(), consume_cancel, reuse, )); let hold = Hold::acquire(&directory).unwrap(); let path = directory.join("read"); fs::write(&path, b"hello").unwrap(); let file = OpenOptions::new().read(true).open(path).unwrap(); let probe = OpenOptions::new() .read(true) .write(true) .open(directory.join(".hold")) .unwrap(); let id = harness.admit( Request::ReadAt(ReadAtRequest::new( Shared::detached(file, hold), 0, IoBufMut::zeroed(5), Cache::Enabled, )), 0, ); // Complete the real read in the kernel, retaining its CQE and // userspace owner while a later cancellation enters the SQ. assert!( !harness .driver .state .stage_ready_requests(&mut harness.driver.ring.submission()) ); harness.driver.ring.submit_and_wait(1).unwrap(); assert!(matches!( probe.try_lock(), Err(fs::TryLockError::WouldBlock) )); harness.driver.orphan(id, &mut harness.deferred); assert!( !harness .driver .state .stage_cancellations(&mut harness.driver.ring.submission()) ); assert_eq!(harness.driver.ring.submission().len(), 1); // Only processing the target CQE permits resource retirement. // The deferred owner still protects the directory until disposal. harness .driver .state .reap(&mut harness.driver.ring, &mut harness.deferred); assert!(matches!( probe.try_lock(), Err(fs::TryLockError::WouldBlock) )); harness.collect(); let RequestOutput::ReadAt(Ok(buf)) = &harness.completed[0].output else { panic!("real read did not complete successfully"); }; assert_eq!(buf.as_ref(), b"hello"); probe.try_lock().unwrap(); probe.unlock().unwrap(); if consume_cancel { // Leave the real cancellation acknowledgement unobserved. harness.driver.ring.submit_and_wait(1).unwrap(); assert!(harness.driver.ring.submission().is_empty()); assert!(!harness.driver.ring.completion().is_empty()); } else { assert_eq!(harness.driver.ring.submission().len(), 1); } if reuse { let (left, mut right) = UnixStream::pair().unwrap(); right.write_all(b"x").unwrap(); let next = harness.admit(recv(left, 1, None), 1); assert_eq!(next.0.index, id.0.index); assert_ne!(next, id); harness.until(2); assert_eq!(received(&harness.completed[1]), b"x"); } harness.drain(); drop(harness); probe.try_lock().unwrap(); drop(probe); fs::remove_dir_all(directory).unwrap(); } } } #[test] fn test_timeout_between_partial_completions() { let mut harness = Harness::new(1); let (left, _right) = UnixStream::pair().unwrap(); let id = harness.stage(recv(left, 8, None), Some(1), 0); // Partial progress leaves the request queued, with no SQE in flight. harness.simulated_completion(id, 4); assert_eq!(harness.driver.state.ready_queue.len(), 1); // Expiring between SQEs must finish locally, without an async cancel. harness.driver.state.advance_timeouts( harness.start + Duration::from_millis(5), &mut harness.deferred, ); assert!(harness.driver.state.pending_cancels.is_empty()); assert!(harness.driver.is_empty()); harness.collect(); assert!(matches!( harness.completed[0].output, RequestOutput::Recv(Err((_, Error::Timeout))) )); harness.drain(); } #[test] fn test_orphaned_follow_up_cannot_stage_reused_slot() { let mut harness = Harness::new(1); let (left, _right) = UnixStream::pair().unwrap(); let id = harness.stage(recv(left, 8, None), None, 0); harness.simulated_completion(id, 4); harness.orphan(id); // Reuse the slot while the abandoned follow-up ID is still queued. let (left, mut right) = UnixStream::pair().unwrap(); right.write_all(b"fresh").unwrap(); let new = harness.admit(recv(left, 5, None), 1); assert_eq!(new.0.index, id.0.index); harness.until(2); assert_eq!(received(&harness.completed[1]), b"fresh"); harness.drain(); } #[test] fn test_orphaned_read_retries_do_not_requeue() { let directory = std::env::temp_dir().join(format!( "commonware_driver_read_test_{}", std::process::id() )); let hold = Hold::acquire(&directory).unwrap(); let file = OpenOptions::new() .read(true) .write(true) .create(true) .truncate(true) .open(directory.join("read")) .unwrap(); let held = Shared::detached(file, hold); for close in [false, true] { for result in [-libc::EAGAIN, 2] { let mut harness = Harness::new(1); let id = harness.stage( Request::ReadAt(ReadAtRequest::new( held.clone(), 0, IoBufMut::zeroed(5), Cache::Enabled, )), None, 0, ); if close { harness.close(); } else { harness.orphan(id); } // ReadAt reports a retry or partial progress even after // cancellation. CancelRequested prevents a follow-up SQE // and retires the request at this CQE. harness.simulated_completion(id, result); assert!(harness.driver.state.ready_queue.is_empty()); assert!(harness.driver.is_empty()); assert_eq!(harness.completed.len(), 1); assert!(matches!( harness.completed[0].observer, TestObserver::Orphaned )); harness.drain(); } } drop(held); fs::remove_dir_all(directory).unwrap(); } /// Cancelling a direct read never stages a successor. When its observer is dropped or its /// worker closes, a read the kernel holds keeps its open and buffer until its operation CQE /// decides its outcome, even when the cancellation acknowledgement arrives first. A read the /// kernel does not hold releases them at once when its observer is dropped, whether it is /// queued for its first submission or requeued after aligned progress or a switch to a /// buffered read. #[test] fn test_cancelled_direct_reads_retain_their_owners_until_completion() { let (directory, file) = make_file("direct_cancel"); let fd = file.set_direct( OpenOptions::new() .read(true) .open(directory.join("blob")) .unwrap(), ); let probe = OpenOptions::new() .write(true) .open(directory.join(".hold")) .unwrap(); // Each read covers three blocks less 200 bytes at offset 0, so its superset is three // blocks. An output returns its request's buffer, so the buffer's address ties the two. let len = 3 * DIRECT_ALIGN - 200; let request = || { let mut request = ReadAtRequest::direct(file.clone(), fd, 0, len).unwrap(); let target = request.buf.as_mut_ptr(); (Request::ReadAt(request), target) }; let returned = |completed: &mut Completed| match &mut completed.output { RequestOutput::ReadAt(Ok(buf) | Err((buf, _))) => buf.as_mut_ptr(), _ => unreachable!("a read request must return a read output"), }; for close in [false, true] { for acknowledged in [false, true] { for (result, outcome) in [ (-libc::EAGAIN, Err(Error::Timeout)), (DIRECT_ALIGN as i32, Err(Error::Timeout)), (10, Err(Error::Timeout)), (0, Err(Error::Timeout)), (-libc::EINVAL, Err(Error::Timeout)), (-libc::EIO, Err(Error::Timeout)), (-libc::ECANCELED, Err(Error::ReadFailed)), (3 * DIRECT_ALIGN as i32, Ok(())), ] { // Orphan a staged read or close its worker, with or without a simulated // cancellation acknowledgement. The kernel may write the buffer until the // operation CQE, so the request keeps its open and buffer. let mut harness = Harness::new(1); let (request, target) = request(); let id = harness.stage(request, None, 0); if close { harness.close(); } else { harness.orphan(id); } if acknowledged { harness.driver.state.handle_cqe( CANCEL_USER_DATA, 0, 0, &mut harness.deferred, ); harness.collect(); } assert_eq!(Arc::strong_count(&file), 2); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(harness.completed.is_empty()); // The simulated operation CQE ends the read without a successor. A covering // completion succeeds, ECANCELED fails, and any other result times out, // including one that would otherwise continue the read or fall back. harness.simulated_completion(id, result); assert!(harness.driver.is_empty()); assert!(harness.driver.state.ready_queue.is_empty()); assert_eq!(harness.driver.state.waiters.in_flight(), 0); let RequestOutput::ReadAt(output) = &harness.completed[0].output else { panic!("a read request must return a read output"); }; match (output, outcome) { (Ok(buf), Ok(())) => assert_eq!(buf.len(), len), (Err((_, Error::Timeout)), Err(Error::Timeout)) | (Err((_, Error::ReadFailed)), Err(Error::ReadFailed)) => {} (output, outcome) => panic!("{result}: {output:?} must match {outcome:?}"), } // The CQE released the open, and the unobserved output owns the buffer until // the worker disposes of it. assert_eq!(Arc::strong_count(&file), 1); assert_eq!(harness.completed.len(), 1); assert_eq!(returned(&mut harness.completed[0]), target); harness.completed.clear(); harness.drain(); } } } // Cancel a read still queued for its first submission, or requeued after simulated // aligned progress or a rejection that switched it to a buffered read. The kernel holds // none of them, so cancelling completes the read at once with a timeout and releases its // open and buffer, and the next service has nothing to submit. for progress in [None, Some(DIRECT_ALIGN as i32), Some(-libc::EINVAL)] { let mut harness = Harness::new(1); let (request, target) = request(); let id = match progress { None => harness.admit(request, 0), Some(result) => { let id = harness.stage(request, None, 0); harness.simulated_completion(id, result); assert_eq!(harness.driver.state.ready_queue.len(), 1); id } }; assert_eq!(harness.driver.state.waiters.in_flight(), 0); assert_eq!(Arc::strong_count(&file), 2); harness.orphan(id); assert!(harness.driver.is_empty()); assert_eq!(Arc::strong_count(&file), 1); assert_eq!(harness.completed.len(), 1); assert_eq!(returned(&mut harness.completed[0]), target); assert!(matches!( harness.completed[0].output, RequestOutput::ReadAt(Err((_, Error::Timeout))) )); harness.service(); assert!(harness.driver.state.ready_queue.is_empty()); assert_eq!(harness.driver.state.waiters.in_flight(), 0); harness.completed.clear(); harness.drain(); } // Only the open retains the directory hold, so dropping it releases the hold. assert!(matches!( probe.try_lock(), Err(fs::TryLockError::WouldBlock) )); drop(file); probe.try_lock().unwrap(); drop(probe); fs::remove_dir_all(directory).unwrap(); } /// A direct read whose observer is forwarded before its rejection falls back within the same /// request and delivers the buffered result through that observer. #[test] fn test_forwarded_direct_read_falls_back_through_its_observer() { let (directory, file) = make_file("direct_forwarded"); let data: Vec = (0..2 * DIRECT_ALIGN).map(|i| (i % 251) as u8).collect(); fs::write(directory.join("blob"), &data).unwrap(); let fd = file.set_direct( OpenOptions::new() .read(true) .open(directory.join("blob")) .unwrap(), ); let request = Request::ReadAt(ReadAtRequest::direct(file.clone(), fd, 100, DIRECT_ALIGN).unwrap()); // Forward the observer of a staged read, then simulate a rejection of its direct read. // The request switches to a buffered read and requeues, so it stays in the driver and the // forwarded observer has no result yet. let mut harness = Harness::new(1); let id = harness.stage(request, None, 0); let (sender, mut receiver) = oneshot::channel(); harness.driver.forward(id, sender, &mut harness.deferred); harness.collect(); harness.simulated_completion(id, -libc::EINVAL); assert_eq!(harness.driver.len(), 1); assert_eq!(harness.driver.state.ready_queue.len(), 1); // The kernel serves the buffered read through the open's file, and the request retires // with one result for the forwarded observer. let limit = Instant::now() + Duration::from_secs(10); while !harness.driver.is_empty() { assert!(Instant::now() < limit, "the buffered read must finish"); harness.service(); } assert_eq!(harness.deferred.results.len(), 1); // Publishing the result delivers the range's bytes to the forwarded observer. for (sender, output) in harness.deferred.results.drain(..) { let _ = sender.send(output); } let Ok(Ok(RequestOutput::ReadAt(Ok(buf)))) = receiver.try_recv() else { panic!("the forwarded observer must receive a successful read"); }; assert_eq!(buf.as_ref(), &data[100..100 + DIRECT_ALIGN]); harness.drain(); drop(file); fs::remove_dir_all(directory).unwrap(); } /// The kernel serves direct reads through a real `O_DIRECT` descriptor, for an unaligned /// range and for a range ending at an unaligned end of file. The open's own file is empty, so /// only a read served by the direct descriptor returns the data. #[test] fn test_direct_reads_are_served_by_the_direct_descriptor() { let (directory, file) = make_file("direct_served"); let data: Vec = (0..3 * DIRECT_ALIGN - 50) .map(|i| (i % 251) as u8) .collect(); fs::write(directory.join("data"), &data).unwrap(); let fd = file.set_direct( OpenOptions::new() .read(true) .custom_flags(libc::O_DIRECT) .open(directory.join("data")) .expect("temporary directory must support O_DIRECT"), ); // The direct read of the first range fills its whole superset [0, 8192). The second range // ends at the end of the file, inside the last superset, so its direct read completes // short and unaligned. let mut harness = Harness::new(1); for (offset, len) in [(100, 5000), (2 * DIRECT_ALIGN + 10, DIRECT_ALIGN - 60)] { let request = ReadAtRequest::direct(file.clone(), fd, offset as u64, len).unwrap(); harness.admit(Request::ReadAt(request), 0); harness.until(1); let RequestOutput::ReadAt(Ok(buf)) = harness.completed.remove(0).output else { panic!("the direct read of {offset}..{} must succeed", offset + len); }; assert_eq!(buf.as_ref(), &data[offset..offset + len]); } harness.drain(); drop(file); fs::remove_dir_all(directory).unwrap(); } #[test] fn test_orphan_in_flight_cancels_once_and_releases_deadline() { for expire_bucket in [false, true] { let mut harness = Harness::new(1); let deadline = harness.start + Duration::from_secs(10); let (left, _right) = UnixStream::pair().unwrap(); let id = harness.admit(recv(left, 8, Some(deadline)), 0); let (survivor, _survivor_peer) = UnixStream::pair().unwrap(); if expire_bucket { harness.admit(recv(survivor, 1, Some(deadline)), 1); } harness.service(); assert!(harness.driver.state.waiters.is_in_flight(id)); // Repeated observer drops must not queue another cancel or remove the // same deadline from the wheel twice. harness.orphan(id); harness.orphan(id); if expire_bucket { // A live sibling forces the orphan's stale wheel record through // expiry before its cancellation or operation CQE arrives. harness .driver .state .advance_timeouts(deadline, &mut harness.deferred); assert!(harness.driver.state.waiters.is_in_flight(id)); } assert_eq!(harness.driver.state.pending_cancels.len(), 1); assert!(harness.driver.next_deadline().is_none()); harness.collect(); harness.drain(); assert_eq!(harness.completed.len(), 1 + usize::from(expire_bucket)); assert_eq!( harness .completed .iter() .filter(|completed| matches!(completed.observer, TestObserver::Orphaned)) .count(), 1 ); if expire_bucket { let survivor = harness .completed .iter() .find(|completed| matches!(completed.observer, TestObserver::Local(1))) .unwrap(); assert!(matches!( survivor.output, RequestOutput::Recv(Err((_, Error::Timeout))) )); } } } #[test] fn test_expiring_bucket_preserves_unconsumed_success() { let mut harness = Harness::new(1); let deadline = harness.start + Duration::from_secs(30); let (socket, mut peer) = UnixStream::pair().unwrap(); peer.write_all(b"x").unwrap(); let completed = harness.admit(recv(socket, 1, Some(deadline)), 0); let (survivor, _peer) = UnixStream::pair().unwrap(); harness.admit(recv(survivor, 1, Some(deadline)), 1); // Retain the first result while another request keeps its bucket live. let limit = Instant::now() + Duration::from_secs(10); while harness.driver.state.waiters.is_pending(completed) { assert!(Instant::now() < limit, "receive stalled"); harness .driver .service(Instant::now(), false, &mut harness.deferred) .unwrap(); thread::yield_now(); } harness .driver .state .advance_timeouts(deadline, &mut harness.deferred); assert!(harness.driver.next_deadline().is_none()); harness.collect(); assert!(matches!( harness.completed[0].observer, TestObserver::Local(0) )); assert_eq!(received(&harness.completed[0]), b"x"); harness.drain(); assert_eq!(harness.completed.len(), 2); assert!(matches!( harness.completed[1].observer, TestObserver::Local(1) )); assert!(matches!( harness.completed[1].output, RequestOutput::Recv(Err((_, Error::Timeout))) )); } #[test] fn test_stale_waiter_identity_cannot_orphan_reused_slot() { let mut harness = Harness::new(1); let (old, _old_peer) = UnixStream::pair().unwrap(); let old = harness.admit(recv(old, 1, None), 0); harness.orphan(old); harness.completed.clear(); let (socket, mut peer) = UnixStream::pair().unwrap(); let current = harness.admit(recv(socket, 1, None), 1); assert_eq!(old.0.index, current.0.index); assert_ne!(old, current); // A delayed drop from the old observer cannot affect its replacement. harness.orphan(old); assert_eq!(harness.driver.len(), 1); assert!(harness.completed.is_empty()); peer.write_all(b"x").unwrap(); harness.until(1); assert!(matches!( harness.completed[0].observer, TestObserver::Local(1) )); harness.drain(); } #[test] fn test_deferred_taskrun_is_serviced_during_busy_turn() { // Both SQEs fit, so staging does not itself force a submission syscall. let mut harness = Harness::new(2); let (left, mut right) = UnixStream::pair().unwrap(); harness.admit(recv(left, 1, None), 0); harness.service_at(harness.start, true); assert!(harness.completed.is_empty()); assert!(!harness.driver.is_empty()); assert_eq!(harness.driver.ring.submission().len(), 2); // The next busy turn submits the receive, which then waits for data. harness.service(); assert_eq!(harness.driver.ring.submission().len(), 0); assert!(harness.completed.is_empty()); // The data's completion waits in deferred task work. Only busy turns // run from here, so their enter alone must post it. right.write_all(b"x").unwrap(); harness.until(1); assert_eq!(received(&harness.completed[0]), b"x"); harness.drain(); } #[test] fn test_wake_poll_alone_does_not_require_kernel_service() { let mut harness = Harness::new(1); // Staging the mailbox poll must not make a CPU-only turn enter the kernel. harness.service(); assert!(harness.driver.is_empty()); assert!(!harness.driver.has_pending_submissions()); assert_eq!(harness.driver.ring.submission().len(), 1); harness.drain(); } #[test] fn test_skipped_park_keeps_deferred_service_obligation() { let mut harness = Harness::new(2); let (left, mut right) = UnixStream::pair().unwrap(); right.write_all(b"x").unwrap(); harness.admit(recv(left, 1, None), 0); harness.service_at(harness.start, true); // Publication prevents parking, so the deferred GETEVENTS enter must // still happen through service before the next busy turn. if harness.driver.state.waker.publish() { harness.driver.state.waker.wake(); } assert!(!harness.driver.park(0, None).unwrap()); assert!(!harness.driver.is_empty()); harness.until(1); harness.drain(); } #[test] fn test_publish_wakes_eventfd_blocked_loop() { let mut harness = Harness::new(2); let (left, _right) = UnixStream::pair().unwrap(); harness.admit(recv(left, 1, None), 0); harness.service(); // Publish after arming so the eventfd path must carry the notification. let deadline = Instant::now() + Duration::from_secs(10); let waker = harness.driver.state.waker.clone(); let producer = thread::spawn(move || { wait_until_eventfd_armed(&waker, deadline); if waker.publish() { waker.wake(); } }); assert!(harness.driver.park(0, Some(deadline)).unwrap()); producer.join().unwrap(); assert!(harness.driver.state.waker.pending(0)); // A published sequence alone does not prove eventfd reached the ring. harness.reap_wake(); harness.drain(); } #[test] fn test_park_rearms_wake_poll_with_full_sq() { for stalled in [false, true] { let mut harness = Harness::new(1); let (socket, mut peer) = UnixStream::pair().unwrap(); let id = harness.admit(recv(socket, 1, None), 0); // The receive fills the SQ before wake polling can be installed. // Keep the SQE unsubmitted so park must flush before rearming. assert!( harness .driver .state .fill_submission_queue(&mut harness.driver.ring) ); assert!(harness.driver.ring.submission().is_full()); assert!(harness.driver.state.waiters.is_in_flight(id)); assert!(harness.driver.state.wake_rearm_needed); let _fault = stalled.then(|| inject(&harness.driver, Fault::StallFlush)); if stalled { // An unchanged SQ cannot accept the wake poll. Park must return // to service with the request and rearm still pending. assert!(!harness.driver.park(0, Some(Instant::now())).unwrap()); assert!(harness.driver.state.wake_rearm_needed); assert!(!harness.driver.is_empty()); assert!(harness.driver.ring.submission().is_full()); harness.service(); assert!(!harness.driver.state.wake_rearm_needed); } // Keep the receive blocked. Only a wake published after arming can // produce a CQE during this bounded wait. let deadline = Instant::now() + Duration::from_secs(10); let waker = harness.driver.state.waker.clone(); let producer = thread::spawn(move || { wait_until_eventfd_armed(&waker, deadline); if waker.publish() { waker.wake(); } }); assert!(harness.driver.park(0, Some(deadline)).unwrap()); producer.join().unwrap(); assert!(harness.driver.state.waker.pending(0)); assert!(!harness.driver.state.wake_rearm_needed); // Reap directly so service cannot hide a missing poll by installing // one after park returns. The receive must still be in flight. harness.reap_wake(); assert!(harness.driver.state.waiters.is_in_flight(id)); // Rearming must preserve the receive that occupied the SQ. peer.write_all(b"x").unwrap(); harness.until(1); assert_eq!(received(&harness.completed[0]), b"x"); harness.drain(); } } #[test] fn test_transient_flush_without_sq_progress_returns_through_service() { let mut harness = Harness::new(1); harness.service(); assert_eq!(harness.driver.ring.submission().len(), 1); // Stall the flush needed to get past the already staged mailbox poll. let _fault = inject(&harness.driver, Fault::StallFlush); let (left, mut right) = UnixStream::pair().unwrap(); right.write_all(b"x").unwrap(); let id = harness.admit(recv(left, 1, None), 0); harness.service(); // The failed flush must not consume or mark the queued operation as // in-flight. Its next attempt follows one GETEVENTS service point. assert!(!harness.driver.state.waiters.is_in_flight(id)); assert_eq!(harness.driver.state.ready_queue.front(), Some(&id)); assert!(FAULT.get().is_none()); harness.until(1); assert_eq!(received(&harness.completed[0]), b"x"); harness.drain(); } #[test] fn test_partial_completion_releases_capacity_to_next_queued_request() { let mut harness = Harness::new(1); let (first, mut first_peer) = UnixStream::pair().unwrap(); first_peer.write_all(b"a").unwrap(); harness.admit(recv(first, 2, None), 0); let (second, mut second_peer) = UnixStream::pair().unwrap(); second_peer.write_all(b"x").unwrap(); harness.admit(recv(second, 1, None), 1); // The first request needs another SQE. Its partial CQE releases the // sole slot so the older queued SQE can run before that follow-up. harness.until(1); assert!(matches!( harness.completed[0].observer, TestObserver::Local(1) )); assert_eq!(received(&harness.completed[0]), b"x"); assert!(harness.driver.state.waiters.in_flight() <= 1); // The retry must append to the original buffer and finish second. first_peer.write_all(b"b").unwrap(); harness.until(2); assert!(matches!( harness.completed[1].observer, TestObserver::Local(0) )); assert_eq!(received(&harness.completed[1]), b"ab"); assert_eq!(harness.driver.state.waiters.in_flight(), 0); harness.drain(); } #[test] fn test_same_open_sync_waits_for_failure_accounting() { for state in [ None, Some(WriteAtState::WritingSync), Some(WriteAtState::WritingBeforeSync), ] { let (directory, held) = make_file("sync_order"); held.tracker.write(); held.tracker.complete(); // Both barriers are accepted while staging capacity remains available. let mut harness = Harness::new(2); let first = harness.admit(Request::Sync(SyncRequest::new(held.clone())), 0); let successor = state.map_or_else( || Request::Sync(SyncRequest::new(held.clone())), |state| { let bufs = if state == WriteAtState::WritingBeforeSync { (0..IOVEC_BATCH_SIZE + 1) .map(|_| IoBuf::from(b"x")) .collect() } else { IoBufs::from(IoBuf::from(b"new")) }; Request::WriteAt(WriteAtRequest::new( held.clone(), 0, bufs.into(), state, Cache::Enabled, )) }, ); let second = harness.admit(successor, 1); assert_eq!(harness.driver.len(), 2); assert_eq!(harness.driver.state.ready_queue.pop_front(), Some(first)); assert!( !harness.driver.state.ready_queue.contains(&second), "durability request became ready before its predecessor was accounted" ); // Keep the SQE out of the ring so its failure is supplied only by the // synthetic CQE. This models terminal ordering without a disk fault. harness.driver.state.waiters.stage(first); assert_eq!(harness.driver.state.waiters.in_flight(), 1); assert!(!harness.driver.has_pending_submissions()); harness.driver.state.handle_cqe( first.user_data(), -libc::EIO, 0, &mut harness.deferred, ); assert!(matches!( held.tracker.failure(), Some(Error::BlobSyncFailed(_, _, error)) if error.raw_os_error() == Some(libc::EIO) )); // The already accepted successor observes the retained failure without // staging a second kernel barrier or consuming the first result. harness .driver .service(Instant::now(), true, &mut harness.deferred) .unwrap(); assert!(harness.driver.is_empty()); assert_eq!(harness.driver.state.waiters.in_flight(), 0); harness.collect(); assert_eq!(harness.completed.len(), 2); assert!(matches!( &harness.completed[0].output, RequestOutput::Sync(Err(Error::Io(error))) if error.raw_os_error() == Some(libc::EIO) )); assert!(matches!( &harness.completed[1].output, RequestOutput::Sync(Err(Error::BlobSyncFailed(_, _, error))) | RequestOutput::WriteAt(Err(Error::BlobSyncFailed(_, _, error))) if error.raw_os_error() == Some(libc::EIO) )); harness.drain(); drop(harness); drop(held); fs::remove_dir_all(directory).unwrap(); } } #[test] fn test_durable_write_keeps_permit_through_follow_up_sqes() { for state in [WriteAtState::WritingSync, WriteAtState::WritingBeforeSync] { let multibatch = state == WriteAtState::WritingBeforeSync; let (directory, file) = make_file("durable_follow_up"); file.tracker.write(); file.tracker.complete(); let mut harness = Harness::new(2); let (bufs, completions) = if !multibatch { ( IoBufs::from(IoBuf::from(b"hello")), vec![-libc::EOPNOTSUPP, -libc::EINTR, 2, 3], ) } else { ( (0..IOVEC_BATCH_SIZE + 1) .map(|_| IoBuf::from(b"x")) .collect(), vec![ -libc::EOPNOTSUPP, -libc::EAGAIN, IOVEC_BATCH_SIZE as i32, 1, 0, ], ) }; let first = harness.driver.admit( Request::WriteAt(WriteAtRequest::new( file.clone(), 0, bufs.into(), state, Cache::Disabled, )), Observer::Local(None), harness.start, &mut harness.deferred, ); let second = harness.admit(Request::Sync(SyncRequest::new(file.clone())), 1); // Orphaned work and unconsumed local outputs have the same permit lifetime. if multibatch { harness.driver.orphan(first, &mut harness.deferred); } for (index, result) in completions.iter().enumerate() { assert_eq!(harness.driver.state.ready_queue.pop_front(), Some(first)); assert!(harness.driver.state.ready_queue.is_empty()); harness.driver.state.waiters.stage(first); harness.driver.state.handle_cqe( first.user_data(), *result, 0, &mut harness.deferred, ); harness .driver .state .poll_acquisitions(&mut harness.deferred); if index + 1 < completions.len() { assert_eq!(harness.driver.state.acquiring.len(), 1); assert!(!harness.driver.state.ready_queue.contains(&second)); assert!(file.durability.try_lock().is_err()); } } assert!(harness.driver.state.acquiring.is_empty()); assert!(!harness.driver.state.waiters.is_pending(first)); // A trailing sync credits the write completed before it under the permit, so the // successor resolves without an SQE. A fused write covers only its own range. if multibatch { assert!(harness.driver.state.ready_queue.is_empty()); assert!(!harness.driver.state.waiters.is_pending(second)); harness.collect(); } else { assert_eq!(harness.driver.state.ready_queue.pop_front(), Some(second)); assert!(matches!( harness.driver.observe(first, TaskWaker::noop()), Observation::Ready(RequestOutput::WriteAt(Ok(()))) )); harness.driver.state.waiters.stage(second); harness.simulated_completion(second, 0); } assert_eq!(harness.completed.len(), 1 + usize::from(multibatch)); assert!(harness.completed.iter().all(|completion| matches!( completion.output, RequestOutput::WriteAt(Ok(())) | RequestOutput::Sync(Ok(())) ))); assert!(!file.tracker.is_dirty()); harness.drain(); drop(harness); drop(file); fs::remove_dir_all(directory).unwrap(); } } #[test] fn test_blocked_durability_does_not_count_as_ready_during_compaction() { let (directory, file) = make_file("blocked_compaction"); file.tracker.write(); file.tracker.complete(); let permit = file.durability.clone().try_lock_owned().unwrap(); let mut harness = Harness::new(1); let (active, mut active_peer) = UnixStream::pair().unwrap(); harness.admit(recv(active, 1, None), 0); harness.service(); let (independent, mut independent_peer) = UnixStream::pair().unwrap(); independent_peer.write_all(b"x").unwrap(); let independent = harness.admit(recv(independent, 1, None), 1); for tag in 2..4 { harness.admit(Request::Sync(SyncRequest::new(file.clone())), tag); } for _ in 0..65 { let (socket, _peer) = UnixStream::pair().unwrap(); let stale = harness.admit(recv(socket, 1, None), 4); harness.orphan(stale); } harness.completed.clear(); harness.service(); assert_eq!(harness.driver.state.ready_queue, [independent]); assert_eq!(harness.driver.state.acquiring.len(), 2); assert_eq!(harness.driver.len(), 4); active_peer.write_all(b"x").unwrap(); harness.until(2); assert_eq!(harness.driver.len(), 2); assert_eq!(harness.driver.state.waiters.in_flight(), 0); assert!(!harness.driver.has_pending_submissions()); assert!(harness.driver.state.ready_queue.is_empty()); drop(permit); harness.until(4); for (tag, completed) in harness.completed.iter().enumerate() { assert!(matches!(completed.observer, TestObserver::Local(id) if id == tag as u64)); } harness.drain(); drop(harness); drop(file); fs::remove_dir_all(directory).unwrap(); } #[tokio::test] async fn test_durability_admitted_after_service_registers_its_wait() { let (directory, file) = make_file("admission_poll"); let permit = file.durability.clone().try_lock_owned().unwrap(); let mut harness = Harness::new(2); harness.service(); // The native driver must progress even once the enclosing task has spent its // cooperative budget. while tokio::task::coop::has_budget_remaining() { tokio::task::consume_budget().await; } // A deferred callback can admit work after service, immediately before parking. harness.admit(Request::Sync(SyncRequest::new(file.clone())), 0); drop(permit); assert!( file.durability.try_lock().is_err(), "accepted wait did not reserve the released permit" ); harness.until(1); assert!(matches!( harness.completed[0].output, RequestOutput::Sync(Ok(())) )); harness.drain(); drop(harness); drop(file); fs::remove_dir_all(directory).unwrap(); } #[test] fn test_foreign_durability_release_wakes_closed_worker_park() { let (directory, file) = make_file("closed_permit_wake"); let permit = file.durability.clone().try_lock_owned().unwrap(); let mailbox = Arc::new(Mailbox::new().unwrap()); let mut driver = Driver::new( &RingConfig { size: 2, timeout_wheel_tick: Duration::from_millis(5), }, Duration::from_secs(60), mailbox.waker.clone(), Instant::now(), ) .unwrap(); let mut deferred = Deferred::default(); let (sender, receiver) = oneshot::channel(); driver.admit( Request::Sync(SyncRequest::new(file.clone())), Observer::DetachedSync(sender), Instant::now(), &mut deferred, ); drop(receiver); assert!(mailbox.close().is_empty()); driver.close(&mut deferred); driver.service(Instant::now(), true, &mut deferred).unwrap(); // Consume a latched acquisition-poll wake before the measured park, then service // again as cleanup does after every wait. No operation SQE can produce a CQE. driver.park(0, Some(Instant::now())).unwrap(); driver.service(Instant::now(), true, &mut deferred).unwrap(); assert_eq!(driver.len(), 1); assert_eq!(driver.state.waiters.in_flight(), 0); assert!(!driver.has_pending_submissions()); assert!(driver.ring.completion().is_empty()); let deadline = Instant::now() + Duration::from_secs(10); let producer = thread::spawn(move || { wait_until_eventfd_armed(&mailbox.waker, deadline); assert!(!mailbox.is_open()); drop(permit); }); assert!(driver.park(0, Some(deadline)).unwrap()); producer.join().unwrap(); let cqe = driver .ring .completion() .next() .expect("permit release did not wake the ring"); assert_eq!(cqe.user_data(), WAKE_USER_DATA); assert!( !driver.state.waker.pending(0), "permit release published an unmatched mailbox batch" ); driver .state .handle_cqe(cqe.user_data(), cqe.result(), cqe.flags(), &mut deferred); driver.service(Instant::now(), true, &mut deferred).unwrap(); assert!(driver.is_empty()); assert_eq!(deferred.completions.len(), 1); let (sender, result) = deferred.completions.pop().unwrap(); assert!(result.is_ok()); assert!(sender.send(result).is_err()); drop(deferred); drop(driver); drop(file); fs::remove_dir_all(directory).unwrap(); } #[test] fn test_shutdown_finishes_orphaned_write_and_detached_sync() { let mut harness = Harness::new(1); let directory = std::env::temp_dir().join(format!("commonware_driver_test_{}", std::process::id())); let hold = Hold::acquire(&directory).unwrap(); let path = directory.join("durable"); let file = OpenOptions::new() .read(true) .write(true) .create(true) .truncate(true) .open(&path) .unwrap(); let held = Shared::detached(file, hold); // More than one iovec batch forces a follow-up write before the sync. let bufs = (0..IOVEC_BATCH_SIZE + 1) .map(|_| IoBuf::from(b"x")) .collect::(); let id = harness.admit( Request::WriteAt(WriteAtRequest::new( held.clone(), 0, bufs.into(), WriteAtState::WritingBeforeSync, Cache::Enabled, )), 0, ); harness.orphan(id); let (sender, receiver) = oneshot::channel(); harness.driver.admit( Request::Sync(SyncRequest::new(held)), Observer::DetachedSync(sender), harness.start, &mut harness.deferred, ); // Retained completion receivers do not participate in drain progress. harness.drain(); assert_eq!(fs::read(&path).unwrap(), vec![b'x'; IOVEC_BATCH_SIZE + 1]); assert_eq!(harness.driver.state.waiters.in_flight(), 0); // Publish only after request ownership has retired. The receiver has // not participated in the drain. let mut published = false; for completed in harness.completed { if let TestObserver::DetachedSync(sender) = completed.observer { let RequestOutput::Sync(result) = completed.output else { panic!("wrong sync result"); }; assert!(sender.send(result).is_ok()); published = true; } } assert!(published); assert!(futures::executor::block_on(receiver).unwrap().is_ok()); fs::remove_dir_all(directory).unwrap(); } /// Voluntary context switches of the calling thread so far. fn voluntary_switches() -> libc::c_long { // SAFETY: rusage holds only integers and timevals, for which all-zero // bytes are a valid value. let mut usage: libc::rusage = unsafe { std::mem::zeroed() }; assert_eq!( // SAFETY: `usage` is valid, writable storage for one rusage. unsafe { libc::getrusage(libc::RUSAGE_THREAD, &mut usage) }, 0 ); usage.ru_nvcsw } /// Busy service turns and parks whose deadline already passed must not /// sleep while a request is unfinished, and such a park must still run the /// deferred task work. A zero-timeout GETEVENTS wait arms a timer and /// schedules whenever no completion is ready, which shows as a voluntary /// context switch per enter. #[test] fn test_busy_service_and_expired_park_do_not_sleep() { /// Turns of one busy service enter and one expired park each. const TURNS: usize = 200; /// Voluntary context switches that unrelated sources may add over the /// turns. A sleeping enter would cost one for every enter. const MAX_VOLUNTARY_SWITCHES: libc::c_long = 20; // Keep a receive in flight with no data, so every turn must enter the // ring for its deferred task work and no completion is ready. let mut harness = Harness::new(2); let (left, mut right) = UnixStream::pair().unwrap(); harness.admit(recv(left, 1, None), 0); harness.service(); assert_eq!(harness.driver.state.waiters.in_flight(), 1); // Each turn makes one busy service enter and one park whose deadline // has already passed. let before = voluntary_switches(); for _ in 0..TURNS { harness.service(); assert!(harness.driver.park(0, Some(Instant::now())).unwrap()); } let switches = voluntary_switches() - before; assert!( switches < MAX_VOLUNTARY_SWITCHES, "{switches} voluntary context switches in {TURNS} turns" ); // The data's completion now waits in deferred task work, which a park // whose deadline already passed must still run. right.write_all(b"x").unwrap(); assert_eq!(harness.driver.ring.completion().len(), 0); assert!(harness.driver.park(0, Some(Instant::now())).unwrap()); assert_eq!(harness.driver.ring.completion().len(), 1); harness.until(1); assert_eq!(received(&harness.completed[0]), b"x"); harness.drain(); } /// A completion the kernel dropped fails every kind of enter with `EBADR`, /// as it fails a wait that finds no completion ready. The kernel checks /// for dropped completions only in that wait. The fault simulates a /// nonzero overflow counter, so this checks how a drop propagates, not /// the counter read itself. #[test] fn test_every_enter_reports_a_dropped_completion() { // A flush fails, here the one park needs before it can rearm wake // polling behind a receive that fills the SQ. let mut harness = Harness::new(1); let (socket, _peer) = UnixStream::pair().unwrap(); harness.admit(recv(socket, 1, None), 0); assert!( harness .driver .state .fill_submission_queue(&mut harness.driver.ring) ); assert!(harness.driver.state.wake_rearm_needed); let fault = inject(&harness.driver, Fault::DropCompletion); let error = harness.driver.park(0, Some(Instant::now())).unwrap_err(); assert_eq!(error.raw_os_error(), Some(libc::EBADR)); // The flush failed before rearming, not a later enter in this park. assert!(harness.driver.state.wake_rearm_needed); drop(fault); harness.drain(); // Keep a receive in flight, so every turn enters the ring. let mut harness = Harness::new(2); let (left, mut right) = UnixStream::pair().unwrap(); harness.admit(recv(left, 1, None), 0); harness.service(); assert_eq!(harness.driver.state.waiters.in_flight(), 1); // A task-work enter fails, here the next busy service turn. let fault = inject(&harness.driver, Fault::DropCompletion); let error = harness .driver .service(Instant::now(), false, &mut harness.deferred) .unwrap_err(); assert_eq!(error.raw_os_error(), Some(libc::EBADR)); drop(fault); // So does a park whose deadline already passed. let fault = inject(&harness.driver, Fault::DropCompletion); let error = harness.driver.park(0, Some(Instant::now())).unwrap_err(); assert_eq!(error.raw_os_error(), Some(libc::EBADR)); drop(fault); // A timed wait fails too, here one that times out. let fault = inject(&harness.driver, Fault::DropCompletion); let deadline = Instant::now() + TIMED_WAIT; let error = harness.driver.park(0, Some(deadline)).unwrap_err(); assert_eq!(error.raw_os_error(), Some(libc::EBADR)); drop(fault); // So does a wait that returns a completion, here an unbounded one // whose receive has data. right.write_all(b"x").unwrap(); let fault = inject(&harness.driver, Fault::DropCompletion); let error = harness.driver.park(0, None).unwrap_err(); assert_eq!(error.raw_os_error(), Some(libc::EBADR)); drop(fault); // With no completion dropped, the receive completes and retires. harness.until(1); assert_eq!(received(&harness.completed[0]), b"x"); harness.drain(); } #[test] fn test_enter_distinguishes_timeout_from_failure() { let mut harness = Harness::new(1); // The empty ring cannot produce a CQE. A wait that times out, a wait // whose deadline already passed and the enters that never wait still // succeed. for enter in [ Enter::Flush, Enter::TaskWork, Enter::Wait { deadline: Some(Instant::now() + TIMED_WAIT), }, Enter::Wait { deadline: Some(Instant::now()), }, ] { Driver::enter(&mut harness.driver.ring, enter).unwrap(); } // SAFETY: This isolated test closes the unused ring descriptor once, // then forgets the ring mapping owner before another FD can reuse it. assert_eq!(unsafe { libc::close(harness.driver.ring.as_raw_fd()) }, 0); let results = [ Enter::Flush, Enter::TaskWork, Enter::Wait { deadline: None }, Enter::Wait { deadline: Some(Instant::now() + Duration::from_secs(1)), }, Enter::Wait { deadline: Some(Instant::now()), }, ] .map(|enter| Driver::enter(&mut harness.driver.ring, enter)); std::mem::forget(harness.driver.ring); // Every enter path must preserve a permanent error. Forget the closed // ring before asserting so a failure cannot try to close its FD again. for result in results { assert_eq!(result.unwrap_err().raw_os_error(), Some(libc::EBADF)); } } }