hyperlight_common/virtq/

producer.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 The Hyperlight Authors.

use alloc::collections::VecDeque;
use alloc::vec::Vec;

use bytes::Bytes;
use smallvec::SmallVec;

use super::*;

/// A used chain observed by the driver (producer) side.
///
/// Read-only chains are returned as [`Ack`](Self::Ack). Chains with a writable
/// buffer complete as [`Data`](Self::Data), even when the device wrote zero
/// bytes. Non-empty segments in [`Data`](Self::Data) are backed by
/// shared-memory pool allocations that are returned when the last clone is
/// dropped.
#[derive(Debug)]
pub enum UsedChain {
    /// Acknowledgement for a read-only/fire-and-forget chain.
    Ack(Token),
    /// Data written by the consumer into the chain's writable buffers.
    ///
    /// The payload may contain zero bytes when the consumer writes zero bytes;
    /// that is still a data used chain because the submitted chain had
    /// writable capacity.
    Data(Token, Segments),
}

impl UsedChain {
    /// Token identifying which submitted chain this used chain corresponds to.
    pub fn token(&self) -> Token {
        match self {
            Self::Ack(token) | Self::Data(token, _) => *token,
        }
    }

    /// Data written by the consumer as contiguous bytes, if this chain has data.
    pub fn to_bytes(&self) -> Option<Bytes> {
        match self {
            Self::Ack(_) => None,
            Self::Data(_, segments) => Some(segments.to_bytes()),
        }
    }

    /// Segments written by the consumer, if this chain has data.
    pub fn segments(&self) -> Option<&Segments> {
        match self {
            Self::Ack(_) => None,
            Self::Data(_, segments) => Some(segments),
        }
    }

    /// Consume the used chain and return written data as contiguous bytes, if present.
    pub fn into_bytes(self) -> Option<Bytes> {
        match self {
            Self::Ack(_) => None,
            Self::Data(_, segments) => Some(segments.into_bytes()),
        }
    }

    /// Consume the used chain and return written segments, if present.
    pub fn into_segments(self) -> Option<Segments> {
        match self {
            Self::Ack(_) => None,
            Self::Data(_, segments) => Some(segments),
        }
    }
}

/// Allocation tracking for an in-flight descriptor chain.
///
/// Descriptor lengths have already been published to the ring, so in-flight
/// state only needs the completion token and allocation ownership for later
/// reclaim.
#[derive(Debug)]
pub(crate) struct Inflight {
    token: Token,
    chain: BufferChain,
}

/// A high-level virtqueue producer (driver side).
///
/// The producer sends chains to the consumer (device), and receives used chains.
/// This is used on the driver/guest side.
///
/// # Threading
///
/// The producer is intended for single-threaded, guest-side use. Reply payloads
/// are exposed as zero-copy [`Bytes`] via [`Bytes::from_owner`](bytes::Bytes::from_owner),
/// which requires the owning pool to be `Send`. Do not move the producer
/// or its replies across threads, and do not instantiate it on the multi-threaded
/// host with those pools.
///
/// # Example
///
/// ```ignore
/// let mut producer = VirtqProducer::new(layout, mem, notifier, pool);
///
/// // Build and submit a chain
/// let mut chain = producer.chain().readable(64).writable(64).build()?;
/// chain.write_all(b"hello")?;
/// let token = producer.submit(chain)?;
///
/// // Later, poll for the used chain
/// if let Some(used) = producer.poll()? {
///     assert_eq!(used.token(), token);
///     match used {
///         UsedChain::Data(_, segments) => println!("Got used chain: {:?}", segments),
///         UsedChain::Ack(_) => println!("Got ack"),
///     }
/// }
/// ```
pub struct VirtqProducer<M, N, P> {
    inner: RingProducer<M>,
    notifier: N,
    pool: P,
    next_token: u32,
    inflight: Vec<Option<Inflight>>,
    pending: VecDeque<UsedChain>,
}

impl<M, N, P> VirtqProducer<M, N, P>
where
    M: MemOps + Clone,
    N: Notifier,
    P: BufferProvider + Clone,
{
    /// Create a new virtqueue producer.
    ///
    /// # Arguments
    ///
    /// * `layout` - Ring memory layout (descriptor table and event suppression addresses)
    /// * `mem` - Memory operations implementation for reading/writing to shared memory
    /// * `notifier` - Callback for notifying the device (consumer) about new chains
    /// * `pool` - Buffer allocator for chain payload and reply data
    pub fn new(layout: Layout, mem: M, notifier: N, pool: P) -> Self {
        let inner = RingProducer::new(layout, mem);
        let ring_len = inner.len();

        Self {
            inner,
            pool,
            notifier,
            next_token: 0,
            inflight: (0..ring_len).map(|_| None).collect(),
            pending: VecDeque::with_capacity(ring_len),
        }
    }

    fn dealloc_elems(
        &self,
        elems: impl IntoIterator<Item = BufferElement>,
    ) -> Result<(), VirtqError> {
        let mut first_err = None;
        for elem in elems {
            if let Err(err) = self.pool.dealloc(elem.addr)
                && first_err.is_none()
            {
                first_err = Some(VirtqError::Alloc(err));
            }
        }

        if let Some(err) = first_err {
            return Err(err);
        }

        Ok(())
    }

    /// Begin building a descriptor chain for submission.
    ///
    /// Returns a [`ChainBuilder`] that allocates buffers from the pool.
    pub fn chain(&self) -> ChainBuilder<M, P> {
        ChainBuilder::new(self.inner.mem().clone(), self.pool.clone())
    }

    /// Begin a batch of submissions.
    ///
    /// Chains submitted through the returned [`SubmitBatch`] are published to
    /// the ring immediately, but the consumer is notified at most once when
    /// [`SubmitBatch::finish`] is called. This mirrors the virtio pattern of
    /// adding multiple buffers and then kicking the queue once.
    pub fn batch(&mut self) -> SubmitBatch<'_, M, N, P> {
        SubmitBatch::new(self)
    }

    /// Submit a [`SendChain`] to the ring.
    ///
    /// Publishes the descriptor chain, stores the in-flight tracking state,
    /// and notifies the consumer if event suppression allows. Notifications
    /// are layout-neutral; use [`batch`](Self::batch) when a higher-level
    /// protocol wants to publish multiple chains and kick once.
    ///
    /// # Errors
    ///
    /// - [`VirtqError::PayloadTooLarge`] - written exceeds readable buffer capacity
    /// - [`VirtqError::RingError`] - ring is full
    /// - [`VirtqError::InvalidState`] - descriptor ID collision
    pub fn submit(&mut self, chain: SendChain<M, P>) -> Result<Token, VirtqError> {
        let cursor_before = self.inner.avail_cursor();
        let token = self.publish(chain)?;
        self.notify_since(cursor_before)?;
        Ok(token)
    }

    fn publish(&mut self, send: SendChain<M, P>) -> Result<Token, VirtqError> {
        let token_id = self.next_token;
        let id = self.inner.submit_available(send.chain())?;
        let token = Token { seq: token_id, id };

        // A free descriptor id must never already be tracked as inflight.
        if self.inflight[id as usize].is_some() {
            return Err(VirtqError::InvalidState);
        }

        let inf = send.into_inflight(token);
        self.inflight[id as usize] = Some(inf);
        self.next_token = self.next_token.wrapping_add(1);

        Ok(token)
    }

    fn notify_since(&mut self, cursor: RingCursor) -> Result<bool, VirtqError> {
        let should_notify = self.inner.should_notify_since(cursor)?;
        if should_notify {
            self.notify_now();
        }
        Ok(should_notify)
    }

    fn notify_now(&self) {
        self.notifier.notify(QueueStats {
            num_free: self.inner.num_free(),
            num_inflight: self.inner.num_inflight(),
        });
    }

    /// Signal backpressure to the consumer.
    ///
    /// Bypasses event suppression. Call this when submit fails with a
    /// backpressure error and the consumer needs to drain.
    #[inline]
    pub fn notify_backpressure(&self) {
        self.notify_now();
    }

    /// Get the current used cursor position.
    ///
    /// Useful for setting up descriptor-based event suppression.
    #[inline]
    pub fn used_cursor(&self) -> RingCursor {
        self.inner.used_cursor()
    }

    /// Number of free (unsubmitted) descriptors in the ring.
    #[inline]
    pub fn num_free(&self) -> usize {
        self.inner.num_free()
    }

    /// Configure event suppression for used buffer notifications.
    ///
    /// This controls when the device (consumer) signals us about completed buffers:
    ///
    /// - [`SuppressionKind::Enable`]: Always signal (default) - good for latency
    /// - [`SuppressionKind::Disable`]: Never signal - caller must poll
    /// - [`SuppressionKind::Descriptor`]: Signal only at specific cursor position
    ///
    /// # Example: Used-chain batching
    ///
    /// ```ignore
    /// // Submit chains, then suppress notifications until all are used
    /// let mut se = producer.chain().readable(64).writable(128).build()?;
    /// se.write_all(b"entry1")?;
    /// producer.submit(se)?;
    /// let cursor = producer.used_cursor();
    /// producer.set_used_suppression(SuppressionKind::Descriptor(cursor))?;
    /// // Device will notify only after reaching that cursor position
    /// ```
    pub fn set_used_suppression(&mut self, kind: SuppressionKind) -> Result<(), VirtqError> {
        match kind {
            SuppressionKind::Enable => self.inner.enable_used_notifications()?,
            SuppressionKind::Disable => self.inner.disable_used_notifications()?,
            SuppressionKind::Descriptor(cursor) => self
                .inner
                .enable_used_notifications_desc(cursor.head(), cursor.wrap())?,
        }
        Ok(())
    }

    /// Reset ring, inflight, and pool state to initial values.
    ///
    /// # Safety
    ///
    /// No outstanding [`UsedChain::Data`] buffers, borrowed segment views, or
    /// peer accesses to previously submitted descriptors may exist. Resetting
    /// recycles the same backing addresses, so outstanding zero-copy buffers or
    /// stale descriptor users could alias memory that is handed out again.
    ///
    /// TODO(virtq): find a way to allow guest to keep used chains across resets.
    pub unsafe fn reset(&mut self) {
        self.inflight.iter_mut().for_each(|slot| *slot = None);
        self.pending.clear();
        self.inner.reset();
        self.pool.reset();
    }

    /// Replace the pool and reset ring, inflight, and pending state.
    ///
    /// # Safety
    ///
    /// No outstanding [`UsedChain::Data`] buffers, borrowed segment views, or
    /// peer accesses to previously submitted descriptors may exist. The new pool
    /// may manage the same shared-memory addresses as the old pool, so old
    /// zero-copy buffers must not outlive this transition.
    pub unsafe fn reset_with_pool(&mut self, pool: P) {
        self.pending.clear();
        self.inflight.iter_mut().for_each(|slot| *slot = None);
        self.inner.reset();
        self.pool = pool;
        self.pool.reset();
    }
}

impl<M, N, P> VirtqProducer<M, N, P>
where
    M: MemOps + Clone + Send + 'static,
    N: Notifier,
    P: BufferProvider + Clone + Send + 'static,
{
    /// Poll for a single used chain from the device.
    ///
    /// Returns buffered used chains from prior [`reclaim`](Self::reclaim)
    /// calls first, then checks the ring for newly used chains.
    ///
    /// Returns `Ok(Some(used))` if a used chain is available, `Ok(None)` if no
    /// used chains are ready (would block), or an error if the device misbehaved.
    ///
    /// Data used chains contain zero-copy [`Bytes`] backed by the shared-memory
    /// allocation via [`BufferOwner`]. The pool allocation is held alive as long
    /// as any `Bytes` clone exists, and is returned to the pool when the last
    /// clone is dropped.
    ///
    /// # Errors
    ///
    /// - [`VirtqError::InvalidState`] - Device returned invalid descriptor ID or
    ///   wrote more data than the writable buffer capacity
    pub fn poll(&mut self) -> Result<Option<UsedChain>, VirtqError> {
        if let Some(chain) = self.pending.pop_front() {
            return Ok(Some(chain));
        }
        self.poll_ring()
    }

    /// Reclaim ring slots and pool allocations from used descriptors.
    ///
    /// Processes all available used chains from the ring: frees readable
    /// buffer allocations immediately, and buffers writable data for
    /// later retrieval via [`poll`](Self::poll).
    ///
    /// Read-only ack used chains are discarded immediately.
    ///
    /// Use this to free resources under backpressure without losing
    /// writable data. Returns the number of chains reclaimed.
    pub fn reclaim(&mut self) -> Result<usize, VirtqError> {
        let mut count = 0;
        while let Some(chain) = self.poll_ring()? {
            if matches!(chain, UsedChain::Data(_, _)) {
                debug_assert!(self.pending.len() < self.inner.len());
                self.pending.push_back(chain);
            }
            count += 1;
        }
        Ok(count)
    }

    /// Poll one used chain directly from the ring (bypassing pending buffer).
    fn poll_ring(&mut self) -> Result<Option<UsedChain>, VirtqError> {
        let used = match self.inner.poll_used() {
            Ok(u) => u,
            Err(RingError::WouldBlock) => return Ok(None),
            Err(e) => return Err(e.into()),
        };

        let inf = self
            .inflight
            .get_mut(used.id as usize)
            .and_then(Option::take)
            .ok_or(VirtqError::InvalidState)?;

        let written = used.len as usize;
        let Inflight { token, chain } = inf;

        self.dealloc_elems(chain.readables().iter().copied())?;

        let used = if chain.writables().is_empty() {
            UsedChain::Ack(token)
        } else {
            UsedChain::Data(token, self.recv_segments(chain.writables(), written)?)
        };

        Ok(Some(used))
    }

    fn recv_segments(
        &self,
        writables: &[BufferElement],
        written: usize,
    ) -> Result<Segments, VirtqError> {
        let mut owned = SmallVec::<[(BufferElement, usize); 4]>::new();
        let mut free = SmallVec::<[BufferElement; 4]>::new();
        let mut remaining = written;

        for &alloc in writables {
            if remaining == 0 {
                free.push(alloc);
                continue;
            }

            let len = remaining.min(alloc.len as usize);
            owned.push((alloc, len));
            remaining -= len;
        }

        if remaining != 0 {
            let elems = owned.iter().map(|(elem, _)| *elem).chain(free);
            self.dealloc_elems(elems)?;
            return Err(VirtqError::InvalidState);
        }

        for (elem, len) in &owned {
            if unsafe { self.inner.mem().as_slice(elem.addr, *len) }.is_err() {
                let elems = owned.iter().map(|(elem, _)| *elem).chain(free);
                let _ = self.dealloc_elems(elems);
                return Err(VirtqError::MemoryReadError);
            }
        }

        let mut sgs = SmallVec::<[Bytes; 4]>::new();
        for (elem, written) in owned {
            let alloc = OwnedAlloc::new(
                self.pool.clone(),
                Allocation {
                    addr: elem.addr,
                    len: elem.len as usize,
                },
            );
            let mem = self.inner.mem().clone();
            let owner = BufferOwner {
                alloc,
                mem,
                written,
            };
            sgs.push(Bytes::from_owner(owner));
        }

        self.dealloc_elems(free)?;

        Ok(Segments::from_smallvec(sgs))
    }

    /// Drain all available used chains, calling the provided closure for each.
    ///
    /// This is a convenience method that repeatedly calls [`poll`](Self::poll)
    /// until no more used chains are available.
    ///
    /// # Arguments
    ///
    /// * `f` - Closure called for each used chain
    ///
    /// # Example
    ///
    /// ```ignore
    /// producer.drain(|used| {
    ///     println!("Got used chain for {:?}", used.token());
    /// })?;
    /// ```
    pub fn drain(&mut self, mut f: impl FnMut(UsedChain)) -> Result<(), VirtqError> {
        while let Some(chain) = self.poll()? {
            f(chain);
        }

        Ok(())
    }
}

/// A scoped batch of producer submissions.
///
/// Submissions are published immediately, while notification is delayed until
/// [`finish`](Self::finish). `finish` is explicit because the event-suppression
/// check can fail; dropping a batch does not notify.
#[must_use = "call finish to notify the consumer about batched submissions"]
pub struct SubmitBatch<'a, M, N, P> {
    producer: &'a mut VirtqProducer<M, N, P>,
    notify_from: Option<RingCursor>,
}

impl<'a, M, N, P> SubmitBatch<'a, M, N, P>
where
    M: MemOps + Clone,
    N: Notifier,
    P: BufferProvider + Clone,
{
    fn new(producer: &'a mut VirtqProducer<M, N, P>) -> Self {
        Self {
            producer,
            notify_from: None,
        }
    }

    /// Begin building a descriptor chain for this batch.
    pub fn chain(&self) -> ChainBuilder<M, P> {
        self.producer.chain()
    }

    /// Publish a chain as part of this batch without notifying yet.
    pub fn submit(&mut self, chain: SendChain<M, P>) -> Result<Token, VirtqError> {
        let cursor_before = self.producer.inner.avail_cursor();
        let token = self.producer.publish(chain)?;

        if self.notify_from.is_none() {
            self.notify_from = Some(cursor_before);
        }
        Ok(token)
    }

    /// Finish the batch and notify the consumer once if event suppression
    /// requires it for the whole published range.
    ///
    /// Returns `true` if a notification was sent.
    pub fn finish(mut self) -> Result<bool, VirtqError> {
        let Some(notify_from) = self.notify_from.take() else {
            return Ok(false);
        };

        self.producer.notify_since(notify_from)
    }
}

/// Builder for configuring a descriptor chain's buffer layout.
///
/// If dropped without building, no resources are leaked (allocations are
/// deferred to [`build`](Self::build)).
#[must_use = "call .build() to create a SendChain"]
pub struct ChainBuilder<M: MemOps, P: BufferProvider + Clone> {
    mem: M,
    pool: P,
    rd_caps: SmallVec<[usize; 4]>,
    wr_caps: SmallVec<[usize; 4]>,
}

impl<M: MemOps, P: BufferProvider + Clone> ChainBuilder<M, P> {
    fn new(mem: M, pool: P) -> Self {
        Self {
            mem,
            pool,
            rd_caps: SmallVec::new(),
            wr_caps: SmallVec::new(),
        }
    }

    /// Request a device-readable buffer of `cap` bytes.
    ///
    /// The producer writes data into readable buffers before submission; the
    /// consumer reads that data after polling the chain.
    /// The actual allocation is deferred to [`build`](Self::build).
    pub fn readable(mut self, cap: usize) -> Self {
        self.rd_caps.push(cap);
        self
    }

    /// Request a device-writable buffer of `cap` bytes.
    ///
    /// The writable buffer is filled by the consumer and returned via
    /// [`VirtqProducer::poll`] as [`UsedChain`].
    ///
    /// Multiple writable buffers are completed as ordered [`Segments`]. The
    /// consumer writes them sequentially, because the virtio used ring reports
    /// one aggregate written length rather than per-descriptor lengths.
    pub fn writable(mut self, cap: usize) -> Self {
        self.wr_caps.push(cap);
        self
    }

    /// Allocate buffers and return a [`SendChain`] for writing.
    ///
    /// # Errors
    ///
    /// - [`VirtqError::InvalidState`] - No buffers requested
    /// - [`VirtqError::Alloc`] - Pool exhausted
    pub fn build(self) -> Result<SendChain<M, P>, VirtqError> {
        if self.rd_caps.is_empty() && self.wr_caps.is_empty() {
            return Err(VirtqError::InvalidState);
        }

        let mut rollback = Rollback::new(&self.pool);
        let mut rd_caps = SmallVec::<[usize; 4]>::new();
        let mut rd_elems = SmallVec::<[BufferElement; 4]>::new();
        let mut wr_elems = SmallVec::<[BufferElement; 4]>::new();

        // Allocate readable buffers, splitting into multiple descriptors if needed.
        // The buffer element lengths are initialized to zero and updated as the
        // `SendChain` writes.
        for &cap in &self.rd_caps {
            let sgs = self.pool.alloc_sg(cap)?;
            let mut remaining = cap;

            for alloc in sgs {
                let _ = checked_descriptor_len(alloc.len)?;
                let seg_cap = remaining.min(alloc.len);

                rd_caps.push(seg_cap);
                rd_elems.push(BufferElement {
                    addr: alloc.addr,
                    len: 0,
                    writable: false,
                });
                remaining -= seg_cap;
                rollback.allocs.push(alloc);
            }

            if remaining != 0 {
                return Err(VirtqError::InvalidState);
            }
        }

        // Allocate writable buffers, with the same caveat about splitting as readable buffers.
        // Writable buffer elements are initialized with their full capacity for the device to
        // write into.
        for &cap in &self.wr_caps {
            let sgs = self.pool.alloc_sg(cap)?;
            for alloc in sgs {
                let len = checked_descriptor_len(alloc.len)?;
                wr_elems.push(BufferElement {
                    addr: alloc.addr,
                    len,
                    writable: true,
                });
                rollback.allocs.push(alloc);
            }
        }

        let chain = BufferChainBuilder::new()
            .readables(rd_elems)
            .writables(wr_elems)
            .build()?;

        rollback.release();

        Ok(SendChain {
            mem: self.mem,
            pool: self.pool,
            chain: Some(chain),
            rd_caps,
            rd_capacity: self.rd_caps.iter().sum(),
            rd_written: 0,
            write_mode: WriteMode::Unset,
        })
    }
}

struct Rollback<'a, P: BufferProvider> {
    pool: &'a P,
    allocs: SmallVec<[Allocation; 8]>,
}

impl<'a, P: BufferProvider> Rollback<'a, P> {
    fn new(pool: &'a P) -> Self {
        Self {
            pool,
            allocs: SmallVec::new(),
        }
    }

    fn release(mut self) {
        self.allocs.clear();
    }
}

impl<P: BufferProvider> Drop for Rollback<'_, P> {
    fn drop(&mut self) {
        for alloc in self.allocs.drain(..) {
            let result = self.pool.dealloc(alloc.addr);
            debug_assert!(result.is_ok(), "rollback dealloc failed: {result:?}");
        }
    }
}

/// Tracks which write API a [`SendChain`] payload uses, so the two paths are
/// not mixed.
///
/// Copy writes ([`SendChain::write`]/[`SendChain::write_all`]) append at an
/// aggregate cursor, while direct writes
/// ([`SendChain::write_seg`]/[`SendChain::with_seg`]) set per-segment lengths
/// absolutely. Mixing them would corrupt the written-length accounting.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WriteMode {
    Unset,
    Append,
    Direct,
}

/// A configured send chain ready for writing and submission.
///
/// Created by [`ChainBuilder::build`]. Write readable payload bytes directly on
/// the chain, then submit via [`VirtqProducer::submit`].
///
/// # Examples
///
/// ```ignore
/// let mut sc = producer.chain().readable(64).writable(128).build()?;
/// sc.write_all(b"header")?.write_all(b" body")?;
/// let tok = producer.submit(sc)?;
///
/// let mut sc = producer.chain().readable(128).build()?;
/// sc.with_seg(0, |buf| serialize_into(buf))?;
/// let tok = producer.submit(sc)?;
/// ```
///
/// Copy writes (`write`/`write_all`) and direct writes (`write_seg`/`with_seg`)
/// must not be mixed on the same chain; doing so panics in debug builds.
///
/// If dropped without submitting, allocated buffers are returned to the pool.
#[must_use = "dropping without submitting deallocates the buffers"]
pub struct SendChain<M: MemOps, P: BufferProvider> {
    mem: M,
    pool: P,
    chain: Option<BufferChain>,
    rd_caps: SmallVec<[usize; 4]>,
    rd_capacity: usize,
    rd_written: usize,
    write_mode: WriteMode,
}

// `chain` is wrapped in `Option` only so `into_inflight` can `take()` it
// without moving out of this `Drop` type; it stays `Some` for a chain's whole
// public lifetime, so these `expect`s cannot fail.
#[allow(clippy::expect_used)]
impl<M: MemOps, P: BufferProvider> SendChain<M, P> {
    fn chain(&self) -> &BufferChain {
        self.chain.as_ref().expect("SendChain missing BufferChain")
    }

    fn chain_mut(&mut self) -> &mut BufferChain {
        self.chain.as_mut().expect("SendChain missing BufferChain")
    }

    /// Record that this chain uses `mode`, asserting it is not mixed with the
    /// other write path.
    fn note_write_mode(&mut self, mode: WriteMode) {
        debug_assert!(
            self.write_mode == WriteMode::Unset || self.write_mode == mode,
            "SendChain mixes copy writes (write/write_all) with direct writes (write_seg/with_seg)"
        );
        self.write_mode = mode;
    }

    fn into_inflight(mut self, token: Token) -> Inflight {
        let chain = self.chain.take().expect("SendChain missing BufferChain");
        Inflight { token, chain }
    }

    /// Number of producer-written readable segments in this chain.
    pub fn segment_count(&self) -> usize {
        self.chain().readables().len()
    }

    /// Total producer-written readable capacity in bytes.
    pub fn capacity(&self) -> usize {
        self.rd_capacity
    }

    /// Number of producer-written readable bytes written so far.
    pub fn written(&self) -> usize {
        self.rd_written
    }

    /// Remaining producer-written readable capacity.
    pub fn remaining(&self) -> usize {
        self.capacity() - self.written()
    }

    /// Write bytes into payload segments, returning how many bytes were written.
    ///
    /// Appends at the current aggregate write position and scatters across
    /// readable segments in chain order. Uses [`MemOps::write`] (volatile on
    /// host side). If `buf` is larger than the remaining capacity, writes as
    /// many bytes as will fit.
    ///
    /// # Errors
    ///
    /// - [`VirtqError::NoPayloadSegment`] - no readable buffer allocated
    /// - [`VirtqError::MemoryWriteError`] - underlying write failed
    pub fn write(&mut self, buf: &[u8]) -> Result<usize, VirtqError> {
        if self.segment_count() == 0 {
            return Err(VirtqError::NoPayloadSegment);
        }

        self.note_write_mode(WriteMode::Append);

        let mut remaining = &buf[..buf.len().min(self.remaining())];
        let mut written = 0;

        let SendChain {
            mem,
            chain,
            rd_caps,
            ..
        } = self;

        let readables = chain
            .as_mut()
            .expect("SendChain missing BufferChain")
            .readables_mut();

        for (readable, &cap) in readables.iter_mut().zip(rd_caps.iter()) {
            if remaining.is_empty() {
                break;
            }

            let written_len = readable.len as usize;
            let free = cap - written_len;
            if free == 0 {
                continue;
            }

            let n = free.min(remaining.len());
            let addr = readable.addr + written_len as u64;
            mem.write(addr, &remaining[..n])
                .map_err(|_| VirtqError::MemoryWriteError)?;

            readable.len += n as u32;
            written += n;
            remaining = &remaining[n..];
        }

        self.rd_written += written;
        Ok(written)
    }

    /// Write the entire buffer into payload segments.
    ///
    /// Appends at the current aggregate write position and scatters across
    /// readable segments in chain order. Uses [`MemOps::write`] (volatile on
    /// host side).
    ///
    /// # Errors
    ///
    /// - [`VirtqError::PayloadTooLarge`] - buf exceeds remaining capacity
    /// - [`VirtqError::NoPayloadSegment`] - no readable buffer allocated
    /// - [`VirtqError::MemoryWriteError`] - underlying write failed
    pub fn write_all(&mut self, buf: &[u8]) -> Result<&mut Self, VirtqError> {
        if self.segment_count() == 0 {
            return Err(VirtqError::NoPayloadSegment);
        }

        if buf.len() > self.remaining() {
            return Err(VirtqError::PayloadTooLarge {
                recv: buf.len(),
                limit: self.remaining(),
            });
        }

        let written = self.write(buf)?;
        debug_assert_eq!(written, buf.len());
        Ok(self)
    }

    /// Write bytes into one readable segment by index.
    ///
    /// Writes from the start of the selected segment and records `buf.len()` as
    /// that segment's descriptor length.
    ///
    /// # Errors
    ///
    /// - [`VirtqError::NoPayloadSegment`] - `index` does not name a payload segment
    /// - [`VirtqError::PayloadTooLarge`] - `buf` exceeds the segment capacity
    /// - [`VirtqError::MemoryWriteError`] - underlying write failed
    pub fn write_seg(&mut self, index: usize, buf: &[u8]) -> Result<&mut Self, VirtqError> {
        self.note_write_mode(WriteMode::Direct);

        let cap = *self
            .rd_caps
            .get(index)
            .ok_or(VirtqError::NoPayloadSegment)?;

        if buf.len() > cap {
            return Err(VirtqError::PayloadTooLarge {
                recv: buf.len(),
                limit: cap,
            });
        }

        let addr = self
            .chain()
            .readables()
            .get(index)
            .ok_or(VirtqError::NoPayloadSegment)?
            .addr;

        self.mem
            .write(addr, buf)
            .map_err(|_| VirtqError::MemoryWriteError)?;

        let previous = self.chain().readables()[index].len as usize;
        self.chain_mut().readables_mut()[index].len = checked_descriptor_len(buf.len())?;
        self.rd_written = self.rd_written - previous + buf.len();
        Ok(self)
    }

    /// Serialize directly into one readable segment by index.
    ///
    /// The closure returns the number of valid bytes it wrote. The written
    /// length for that segment is recorded on success.
    ///
    /// # Errors
    ///
    /// - [`VirtqError::NoPayloadSegment`] - `index` does not name a payload segment
    /// - [`VirtqError::PayloadTooLarge`] - closure reports more bytes than segment capacity
    /// - [`VirtqError::MemoryWriteError`] - the memory backend cannot expose a mutable slice
    pub fn with_seg<E>(
        &mut self,
        index: usize,
        f: impl FnOnce(&mut [u8]) -> Result<usize, E>,
    ) -> Result<&mut Self, E>
    where
        E: From<VirtqError>,
    {
        self.note_write_mode(WriteMode::Direct);

        let cap = *self
            .rd_caps
            .get(index)
            .ok_or_else(|| E::from(VirtqError::NoPayloadSegment))?;

        let addr = self
            .chain()
            .readables()
            .get(index)
            .ok_or_else(|| E::from(VirtqError::NoPayloadSegment))?
            .addr;

        let buf = unsafe {
            self.mem
                .as_mut_slice(addr, cap)
                .map_err(|_| E::from(VirtqError::MemoryWriteError))?
        };

        let written = f(buf)?;
        if written > buf.len() {
            return Err(E::from(VirtqError::PayloadTooLarge {
                recv: written,
                limit: buf.len(),
            }));
        }

        let previous = self.chain().readables()[index].len as usize;
        // SAFETY: index was validated by the earlier get() call, so the readable element exists.
        self.chain_mut().readables_mut()[index].len =
            checked_descriptor_len(written).map_err(E::from)?;
        self.rd_written = self.rd_written - previous + written;

        Ok(self)
    }
}

impl<M: MemOps, P: BufferProvider> Drop for SendChain<M, P> {
    fn drop(&mut self) {
        if let Some(chain) = self.chain.take() {
            for elem in chain.elems() {
                let result = self.pool.dealloc(elem.addr);
                debug_assert!(result.is_ok(), "SendChain drop dealloc failed: {result:?}");
            }
        }
    }
}

fn checked_descriptor_len(len: usize) -> Result<u32, VirtqError> {
    if len > u32::MAX as usize {
        return Err(VirtqError::PayloadTooLarge {
            recv: len,
            limit: u32::MAX as usize,
        });
    }
    Ok(len as u32)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::virtq::ring::tests::{TestMem, make_consumer, make_ring};
    use crate::virtq::test_utils::*;

    fn poll_received<M: MemOps + Clone, N: Notifier>(
        consumer: &mut VirtqConsumer<M, N>,
    ) -> (RecvChain, ReplyChain<M>) {
        consumer.poll(1024).unwrap().unwrap()
    }

    #[derive(Clone)]
    struct NoDirectSliceMem(TestMem);

    // SAFETY: Delegates all non-slice memory operations to TestMem. Direct
    // slices are intentionally unsupported to exercise producer error handling.
    unsafe impl MemOps for NoDirectSliceMem {
        type Error = ();

        fn read(&self, addr: u64, dst: &mut [u8]) -> Result<(), Self::Error> {
            self.0.read(addr, dst).map_err(|e| match e {})
        }

        fn write(&self, addr: u64, src: &[u8]) -> Result<(), Self::Error> {
            self.0.write(addr, src).map_err(|e| match e {})
        }

        fn load_acquire(&self, addr: u64) -> Result<u16, Self::Error> {
            self.0.load_acquire(addr).map_err(|e| match e {})
        }

        fn store_release(&self, addr: u64, val: u16) -> Result<(), Self::Error> {
            self.0.store_release(addr, val).map_err(|e| match e {})
        }

        unsafe fn as_slice(&self, _addr: u64, _len: usize) -> Result<&[u8], Self::Error> {
            Err(())
        }

        unsafe fn as_mut_slice(&self, _addr: u64, _len: usize) -> Result<&mut [u8], Self::Error> {
            Err(())
        }
    }

    #[test]
    fn test_chain_readwrite_build() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().readable(64).writable(128).build().unwrap();
        assert_eq!(se.capacity(), 64);
        assert_eq!(se.written(), 0);
        assert_eq!(se.remaining(), 64);
    }

    #[test]
    fn test_chain_readable_writable_names_build() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().readable(16).writable(32).build().unwrap();
        assert_eq!(se.segment_count(), 1);
        assert_eq!(se.capacity(), 16);
    }

    #[test]
    fn test_chain_multi_readable_write_all_scatters() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer
            .chain()
            .readable(5)
            .readable(6)
            .writable(32)
            .build()
            .unwrap();
        se.write_all(b"hello world").unwrap();

        assert_eq!(se.written(), 11);

        let token = producer.submit(se).unwrap();
        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.token(), token);
        assert_eq!(recv.to_bytes().as_ref(), b"hello world");
        assert_eq!(recv.segments().segment_count(), 2);
        assert_eq!(recv.segments().as_slice()[0].as_ref(), b"hello");
        assert_eq!(recv.segments().as_slice()[1].as_ref(), b" world");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_chain_readable_splits_logical_capacity() {
        let ring = make_ring(16);
        let layout = ring.layout();
        let mem = ring.mem();
        let pool_base = mem.base_addr() + Layout::query_size(ring.len()) as u64 + 0x100;
        let pool = TestPool::new_with_max_alloc_len(pool_base, 0x8000, 4);
        let notifier = TestNotifier::new();
        let mut producer = VirtqProducer::new(layout, mem.clone(), notifier.clone(), pool);
        let mut consumer = VirtqConsumer::new(layout, mem, notifier);

        let mut se = producer.chain().readable(10).writable(32).build().unwrap();

        assert_eq!(se.segment_count(), 3);
        assert_eq!(se.capacity(), 10);

        se.write_all(b"abcdefghij").unwrap();
        assert_eq!(se.written(), 10);

        let token = producer.submit(se).unwrap();
        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.token(), token);
        assert_eq!(recv.to_bytes().as_ref(), b"abcdefghij");
        assert_eq!(recv.segments().segment_count(), 3);
        assert_eq!(recv.segments().as_slice()[0].as_ref(), b"abcd");
        assert_eq!(recv.segments().as_slice()[1].as_ref(), b"efgh");
        assert_eq!(recv.segments().as_slice()[2].as_ref(), b"ij");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_chain_readable_rejects_zero_capacity_on_build() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        assert!(matches!(
            producer.chain().readable(0).build(),
            Err(VirtqError::Alloc(AllocError::InvalidArg))
        ));
    }

    #[test]
    fn test_chain_writable_splits_logical_capacity() {
        let ring = make_ring(16);
        let layout = ring.layout();
        let mem = ring.mem();
        let pool_base = mem.base_addr() + Layout::query_size(ring.len()) as u64 + 0x100;
        let pool = TestPool::new_with_max_alloc_len(pool_base, 0x8000, 4);
        let notifier = TestNotifier::new();
        let mut producer = VirtqProducer::new(layout, mem.clone(), notifier.clone(), pool);
        let mut consumer = VirtqConsumer::new(layout, mem, notifier);

        let se = producer.chain().writable(10).build().unwrap();
        let token = producer.submit(se).unwrap();

        let (_recv, reply) = poll_received(&mut consumer);
        let ReplyChain::Writable(mut wc) = reply else {
            panic!("expected writable reply");
        };
        assert_eq!(wc.capacity(), 10);

        wc.write_all(b"abcdefghij").unwrap();
        consumer.complete(wc).unwrap();

        let used = producer.poll().unwrap().unwrap();
        assert_eq!(used.token(), token);
        let segments = used.segments().unwrap();
        assert_eq!(segments.segment_count(), 3);
        assert_eq!(segments.as_slice()[0].as_ref(), b"abcd");
        assert_eq!(segments.as_slice()[1].as_ref(), b"efgh");
        assert_eq!(segments.as_slice()[2].as_ref(), b"ij");
    }

    #[test]
    fn test_chain_writable_rejects_zero_capacity_on_build() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        assert!(matches!(
            producer.chain().writable(0).build(),
            Err(VirtqError::Alloc(AllocError::InvalidArg))
        ));
    }

    #[test]
    fn test_chain_multi_readable_write_all_preserves_segments() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(4).readable(4).build().unwrap();
        se.write_all(b"headbody").unwrap();

        producer.submit(se).unwrap();
        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"headbody");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_chain_payload_segment_writer_serializes_directly() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(4).readable(4).build().unwrap();
        se.with_seg(0, |segment| {
            segment.copy_from_slice(b"head");
            Ok::<usize, VirtqError>(4)
        })
        .unwrap();
        se.with_seg(1, |segment| {
            segment.copy_from_slice(b"body");
            Ok::<usize, VirtqError>(4)
        })
        .unwrap();

        producer.submit(se).unwrap();
        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"headbody");
        assert_eq!(recv.segments().segment_count(), 2);
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_chain_payload_segment_write_copies_directly() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(4).readable(4).build().unwrap();
        se.write_seg(0, b"head").unwrap();
        se.write_seg(1, b"body").unwrap();

        producer.submit(se).unwrap();
        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"headbody");
        assert_eq!(recv.segments().segment_count(), 2);
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_chain_multi_writable_used_returns_segments() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().writable(5).writable(6).build().unwrap();
        let token = producer.submit(se).unwrap();

        let (_recv, reply) = poll_received(&mut consumer);
        let ReplyChain::Writable(mut wc) = reply else {
            panic!("expected writable reply");
        };
        assert_eq!(wc.capacity(), 11);

        wc.write_all(b"hello world").unwrap();
        consumer.complete(wc).unwrap();

        let used = producer.poll().unwrap().unwrap();
        assert_eq!(used.token(), token);
        let segments = used.segments().unwrap();
        assert_eq!(segments.segment_count(), 2);
        assert_eq!(segments.as_slice()[0].as_ref(), b"hello");
        assert_eq!(segments.as_slice()[1].as_ref(), b" world");
        assert_eq!(segments.to_bytes().as_ref(), b"hello world");
    }

    #[test]
    fn test_chain_multi_writable_short_used_truncates_last_segment() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().writable(5).writable(6).build().unwrap();
        producer.submit(se).unwrap();

        let (_recv, reply) = poll_received(&mut consumer);
        let ReplyChain::Writable(mut wc) = reply else {
            panic!("expected writable reply");
        };

        wc.write_all(b"hello wo").unwrap();
        consumer.complete(wc).unwrap();

        let used = producer.poll().unwrap().unwrap();
        let segments = used.segments().unwrap();
        assert_eq!(segments.segment_count(), 2);
        assert_eq!(segments.as_slice()[0].as_ref(), b"hello");
        assert_eq!(segments.as_slice()[1].as_ref(), b" wo");
        assert_eq!(segments.to_bytes().as_ref(), b"hello wo");
    }

    #[test]
    fn test_chain_multi_writable_zero_used_returns_empty_segments() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().writable(5).writable(6).build().unwrap();
        producer.submit(se).unwrap();

        let (_recv, reply) = poll_received(&mut consumer);
        consumer.complete(reply).unwrap();

        let used = producer.poll().unwrap().unwrap();
        let segments = used.segments().unwrap();
        assert_eq!(segments.segment_count(), 0);
        assert!(segments.is_empty());
        assert!(segments.to_bytes().is_empty());
    }

    #[test]
    fn test_chain_readable_only_build() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().readable(32).build().unwrap();
        assert_eq!(se.capacity(), 32);
    }

    #[test]
    fn test_chain_writable_only_build() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().writable(64).build().unwrap();
        assert_eq!(se.capacity(), 0);
    }

    #[test]
    fn test_chain_empty_build_fails() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let result = producer.chain().build();
        assert!(matches!(result, Err(VirtqError::InvalidState)));
    }

    #[test]
    fn test_send_chain_write_all_and_submit() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();

        se.write_all(b"hello")
            .unwrap()
            .write_all(b" world")
            .unwrap();
        assert_eq!(se.written(), 11);
        assert_eq!(se.remaining(), 53);
        let tok = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.token(), tok);
        assert_eq!(recv.to_bytes().as_ref(), b"hello world");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_send_payload_write_all_fluent() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();
        se.write_all(b"hello")
            .unwrap()
            .write_all(b" world")
            .unwrap();
        assert_eq!(se.written(), 11);
        assert_eq!(se.remaining(), 53);
        let tok = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.token(), tok);
        assert_eq!(recv.to_bytes().as_ref(), b"hello world");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_send_payload_partial_write() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(8).build().unwrap();
        let written = se.write(b"hello world").unwrap();
        assert_eq!(written, 8);
        assert_eq!(se.remaining(), 0);

        producer.submit(se).unwrap();
        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"hello wo");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_send_payload_write_with_serializes_directly() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();
        se.with_seg(0, |buf| {
            buf[..5].copy_from_slice(b"hello");
            Ok::<usize, VirtqError>(5)
        })
        .unwrap();

        let _tok = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"hello");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_send_chain_single_segment_writer_serializes_directly() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();
        se.with_seg(0, |segment| {
            assert_eq!(segment.len(), 64);
            segment[..5].copy_from_slice(b"hello");
            Ok::<usize, VirtqError>(5)
        })
        .unwrap();

        let _tok = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"hello");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_send_chain_single_segment_writer_rejects_multi_segment() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(4).readable(4).build().unwrap();
        assert!(matches!(
            se.with_seg(2, |_| Ok::<usize, VirtqError>(0)),
            Err(VirtqError::NoPayloadSegment)
        ));
    }

    #[test]
    fn test_send_chain_single_segment_writer_rejects_auto_split_chain() {
        let ring = make_ring(16);
        let layout = ring.layout();
        let mem = ring.mem();
        let pool_base = mem.base_addr() + Layout::query_size(ring.len()) as u64 + 0x100;
        let pool = TestPool::new_with_max_alloc_len(pool_base, 0x8000, 4);
        let notifier = TestNotifier::new();
        let producer = VirtqProducer::new(layout, mem, notifier, pool);

        let mut se = producer.chain().readable(8).build().unwrap();
        assert_eq!(se.segment_count(), 2);
        assert!(matches!(
            se.with_seg(2, |_| Ok::<usize, VirtqError>(0)),
            Err(VirtqError::NoPayloadSegment)
        ));
    }

    #[test]
    fn test_send_payload_segment_set_written_too_large() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(32).writable(64).build().unwrap();
        let err = se
            .with_seg(0, |_| Ok::<usize, VirtqError>(64))
            .err()
            .unwrap();
        assert!(matches!(
            err,
            VirtqError::PayloadTooLarge {
                recv: 64,
                limit: 32
            }
        ));
    }

    #[test]
    fn test_send_chain_write_too_large() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(4).build().unwrap();
        let err = se.write_all(b"too long").err().unwrap();
        assert!(matches!(
            err,
            VirtqError::PayloadTooLarge { recv: 8, limit: 4 }
        ));
    }

    #[test]
    fn test_writeonly_has_no_readable_buffer() {
        let ring = make_ring(16);
        let (producer, _consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().writable(32).build().unwrap();
        let err = se.write_all(b"data").err().unwrap();
        assert!(matches!(err, VirtqError::NoPayloadSegment));
    }

    #[test]
    fn test_drop_chain_builder_deallocs() {
        let ring = make_ring(16);
        let (mut producer, _consumer, _notifier) = make_test_producer(&ring);

        {
            let _builder = producer.chain().readable(64).writable(128);
            // dropped without build
        }

        // Ring should still be fully usable
        let se = producer.chain().readable(64).writable(128).build().unwrap();
        let tok = producer.submit(se).unwrap();
        assert!(tok.id < 16);
    }

    #[test]
    fn test_drop_send_chain_deallocs() {
        let ring = make_ring(16);
        let (mut producer, _consumer, _notifier) = make_test_producer(&ring);

        {
            let _se = producer.chain().readable(64).writable(128).build().unwrap();
            // dropped without submit
        }

        // Ring should still be fully usable
        let se = producer.chain().readable(64).writable(128).build().unwrap();
        let tok = producer.submit(se).unwrap();
        assert!(tok.id < 16);
    }

    #[test]
    fn test_submit_notifies() {
        let ring = make_ring(16);
        let (mut producer, _consumer, notifier) = make_test_producer(&ring);

        let initial_count = notifier.notification_count();

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();
        se.write_all(b"hello").unwrap();
        producer.submit(se).unwrap();

        assert!(notifier.notification_count() > initial_count);
    }

    #[test]
    fn test_submit_read_only_notifies_by_default() {
        let ring = make_ring(16);
        let (mut producer, _consumer, notifier) = make_test_producer(&ring);

        let initial_count = notifier.notification_count();

        let mut se = producer.chain().readable(64).build().unwrap();
        se.write_all(b"fire-and-forget").unwrap();
        producer.submit(se).unwrap();

        assert!(notifier.notification_count() > initial_count);
    }

    #[test]
    fn test_submit_write_only_notifies_by_default() {
        let ring = make_ring(16);
        let (mut producer, _consumer, notifier) = make_test_producer(&ring);

        let initial_count = notifier.notification_count();

        let se = producer.chain().writable(128).build().unwrap();
        producer.submit(se).unwrap();

        assert!(notifier.notification_count() > initial_count);
    }

    #[test]
    fn test_batch_notifies_once_on_finish() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, notifier) = make_test_producer(&ring);

        let initial_count = notifier.notification_count();

        let mut batch = producer.batch();

        let mut first = batch.chain().readable(64).build().unwrap();
        first.write_all(b"first").unwrap();
        batch.submit(first).unwrap();

        let mut second = batch.chain().readable(64).build().unwrap();
        second.write_all(b"second").unwrap();
        batch.submit(second).unwrap();

        assert_eq!(notifier.notification_count(), initial_count);
        assert!(batch.finish().unwrap());
        assert_eq!(notifier.notification_count(), initial_count + 1);

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"first");
        consumer.complete(reply).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"second");
        consumer.complete(reply).unwrap();
    }

    #[test]
    fn test_batch_finish_notifies_from_batch_start_cursor() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, notifier) = make_test_producer(&ring);

        let cursor = consumer.avail_cursor();
        consumer
            .set_avail_suppression(SuppressionKind::Descriptor(cursor))
            .unwrap();

        let mut batch = producer.batch();

        let mut first = batch.chain().readable(64).build().unwrap();
        first.write_all(b"first").unwrap();
        batch.submit(first).unwrap();
        assert_eq!(notifier.notification_count(), 0);

        let mut second = batch.chain().readable(64).writable(64).build().unwrap();
        second.write_all(b"second").unwrap();
        batch.submit(second).unwrap();

        assert!(batch.finish().unwrap());

        assert_eq!(notifier.notification_count(), 1);
    }

    #[test]
    fn test_empty_batch_finish_does_not_notify() {
        let ring = make_ring(16);
        let (mut producer, _consumer, notifier) = make_test_producer(&ring);

        let batch = producer.batch();
        assert!(!batch.finish().unwrap());
        assert_eq!(notifier.notification_count(), 0);
    }

    #[test]
    fn test_write_only_round_trip() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().writable(32).build().unwrap();
        let token = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.token(), token);
        assert!(recv.to_bytes().is_empty());

        if let ReplyChain::Writable(mut wc) = reply {
            wc.write_all(b"filled-by-consumer").unwrap();
            consumer.complete(wc).unwrap();
        } else {
            panic!("expected Writable");
        }

        let used = producer.poll().unwrap().unwrap();
        assert_eq!(used.token(), token);
        assert_eq!(used.to_bytes().unwrap().len(), b"filled-by-consumer".len());
        assert_eq!(used.to_bytes().unwrap().as_ref(), b"filled-by-consumer");
    }

    #[test]
    fn test_read_only_round_trip() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(32).build().unwrap();
        se.write_all(b"fire-and-forget").unwrap();
        let token = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.token(), token);
        assert_eq!(recv.to_bytes().as_ref(), b"fire-and-forget");
        assert!(matches!(reply, ReplyChain::Ack(_)));
        consumer.complete(reply).unwrap();

        let used = producer.poll().unwrap().unwrap();
        assert!(matches!(used, UsedChain::Ack(t) if t == token));
    }

    #[test]
    fn test_readwrite_round_trip() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();
        se.write_all(b"request data").unwrap();
        let token = producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"request data");
        if let ReplyChain::Writable(mut wc) = reply {
            wc.write_all(b"response data").unwrap();
            consumer.complete(wc).unwrap();
        } else {
            panic!("expected Writable");
        }

        let used = producer.poll().unwrap().unwrap();
        assert_eq!(used.token(), token);
        assert_eq!(used.to_bytes().unwrap().as_ref(), b"response data");
    }

    #[test]
    fn test_poll_used_requires_direct_slice() {
        let ring = make_ring(16);
        let layout = ring.layout();
        let test_mem = ring.mem();
        let pool_base = test_mem.base_addr() + Layout::query_size(ring.len()) as u64 + 0x100;
        let pool = TestPool::new(pool_base, 0x8000);
        let notifier = TestNotifier::new();
        let mem = NoDirectSliceMem(test_mem);
        let mut producer = VirtqProducer::new(layout, mem.clone(), notifier.clone(), pool);
        let mut consumer = VirtqConsumer::new(layout, mem, notifier);

        let mut se = producer.chain().readable(64).writable(128).build().unwrap();
        se.write_all(b"request data").unwrap();
        producer.submit(se).unwrap();

        let (_recv, reply) = poll_received(&mut consumer);
        if let ReplyChain::Writable(mut wc) = reply {
            wc.write_all(b"response data").unwrap();
            consumer.complete(wc).unwrap();
        } else {
            panic!("expected Writable");
        }

        assert!(matches!(producer.poll(), Err(VirtqError::MemoryReadError)));
    }

    #[test]
    fn test_villain_used_len_exceeding_writable_capacity_is_rejected_and_released() {
        let ring = make_ring(16);
        let (mut producer, _consumer, _notifier) = make_test_producer(&ring);
        let mut ring_consumer = make_consumer(&ring);

        let se = producer.chain().writable(4).build().unwrap();
        producer.submit(se).unwrap();

        let (id, _) = ring_consumer.poll_available().unwrap();
        ring_consumer.submit_used(id, 8).unwrap();

        assert!(matches!(producer.poll(), Err(VirtqError::InvalidState)));
        assert_eq!(producer.inner.num_inflight(), 0);

        let se = producer.chain().writable(4).build().unwrap();
        producer.submit(se).unwrap();
        assert_eq!(producer.inner.num_inflight(), 1);
    }

    #[test]
    fn test_villain_used_descriptor_with_invalid_id_is_rejected() {
        let ring = make_ring(16);
        let (mut producer, _consumer, _notifier) = make_test_producer(&ring);

        let se = producer.chain().writable(4).build().unwrap();
        producer.submit(se).unwrap();

        let mut desc = Descriptor::new(0, 0, ring.len() as u16, DescFlags::empty());
        desc.mark_used(true);
        ring.write_desc(0, desc);

        assert!(matches!(
            producer.poll(),
            Err(VirtqError::RingError(RingError::InvalidState))
        ));
        assert_eq!(producer.inner.num_inflight(), 1);
    }

    #[test]
    fn test_virtq_producer_reset() {
        let ring = make_ring(16);
        let (mut producer, mut consumer, _notifier) = make_test_producer(&ring);

        // Submit and complete a round trip
        let mut se = producer.chain().readable(32).writable(64).build().unwrap();
        se.write_all(b"hello").unwrap();
        producer.submit(se).unwrap();

        let (recv, reply) = poll_received(&mut consumer);
        assert_eq!(recv.to_bytes().as_ref(), b"hello");
        consumer.complete(reply).unwrap();
        let _ = producer.poll().unwrap().unwrap();

        // Now reset
        // SAFETY: the used chain was dropped before reset and no peer can
        // access the reset test ring concurrently.
        unsafe {
            producer.reset();
        }

        // All inflight slots should be cleared
        assert_eq!(producer.inner.num_inflight(), 0);
        // Ring state should be back to initial
        assert_eq!(producer.inner.num_free(), producer.inner.len());
    }

    #[test]
    fn test_virtq_producer_reset_clears_inflight() {
        let ring = make_ring(16);
        let (mut producer, _consumer, _notifier) = make_test_producer(&ring);

        // Submit without completing
        let se = producer.chain().writable(64).build().unwrap();
        producer.submit(se).unwrap();

        assert_eq!(producer.inner.num_inflight(), 1);

        // SAFETY: no peer can access the reset test ring concurrently.
        unsafe {
            producer.reset();
        }

        assert_eq!(producer.inner.num_inflight(), 0);
        assert_eq!(producer.inner.num_free(), producer.inner.len());
    }
}