zircon/kernel/object/socket_dispatcher.rs

// Copyright 2026 The Fuchsia Authors
//
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file or at
// https://opensource.org/licenses/MIT

use super::dispatcher::{
    DispatcherOps, PeerHolder, PeerHolderMuClass, PeeredState, PeeredStateMuGuard,
    impl_peered_dispatcher_facade_with_state,
};
use super::handle::KernelHandle;
use super::mbuf::MBufChain;
use super::socket_dispatcher_ffi::cpp_socket_dispatcher_create;
use crate::counters::define_kcounter;
use crate::user_copy::{UserInPtr, UserOutPtr};
use core::convert::Infallible;
use core::pin::Pin;
use debug::ltrace_entry;
use fbl::{Canary, RefPtr};
use ksync::{KMutex, PhantomMutex, RawCriticalMutex, guarded};
use object_constants_rs::{
    kSocketDispatcherStateAlign, kSocketDispatcherStateOffset, kSocketDispatcherStateSize,
};
use pin_init::{PinInit, pin_data, pin_init, pinned_drop};
use zx_status::Status;
use zx_types::{
    ZX_OBJ_TYPE_SOCKET, ZX_RIGHT_DUPLICATE, ZX_RIGHT_GET_PROPERTY, ZX_RIGHT_INSPECT,
    ZX_RIGHT_MANAGE_SOCKET, ZX_RIGHT_READ, ZX_RIGHT_SET_PROPERTY, ZX_RIGHT_SIGNAL,
    ZX_RIGHT_SIGNAL_PEER, ZX_RIGHT_TRANSFER, ZX_RIGHT_WAIT, ZX_RIGHT_WRITE, ZX_SOCKET_DATAGRAM,
    ZX_SOCKET_DISPOSITION_WRITE_DISABLED, ZX_SOCKET_DISPOSITION_WRITE_ENABLED,
    ZX_SOCKET_PEER_CLOSED, ZX_SOCKET_PEER_WRITE_DISABLED, ZX_SOCKET_READ_THRESHOLD,
    ZX_SOCKET_READABLE, ZX_SOCKET_WRITABLE, ZX_SOCKET_WRITE_DISABLED, ZX_SOCKET_WRITE_THRESHOLD,
    ZX_USER_SIGNAL_ALL, zx_info_socket_t, zx_rights_t, zx_signals_t,
};

pub const DEFAULT_RIGHTS: zx_rights_t = ZX_RIGHT_TRANSFER
    | ZX_RIGHT_DUPLICATE
    | ZX_RIGHT_READ
    | ZX_RIGHT_WRITE
    | ZX_RIGHT_WAIT
    | ZX_RIGHT_INSPECT
    | ZX_RIGHT_GET_PROPERTY
    | ZX_RIGHT_SET_PROPERTY
    | ZX_RIGHT_SIGNAL
    | ZX_RIGHT_SIGNAL_PEER
    | ZX_RIGHT_MANAGE_SOCKET;

pub const ALLOWED_SIGNALS: u32 = ZX_USER_SIGNAL_ALL;

pub const ZX_SOCKET_CREATE_MASK: u32 = ZX_SOCKET_DATAGRAM;

const LOCAL_TRACE: u32 = 0;

zr::static_assert_size_and_align!(
    SocketDispatcherState,
    kSocketDispatcherStateSize,
    kSocketDispatcherStateAlign,
);

define_kcounter!(DISPATCHER_SOCKET_CREATE_COUNT, "dispatcher.socket.create", Sum);
define_kcounter!(DISPATCHER_SOCKET_DESTROY_COUNT, "dispatcher.socket.destroy", Sum);

/// Represents the disposition of a socket endpoint's write status.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[repr(u32)]
pub enum Disposition {
    /// No change in disposition.
    None = 0,
    /// Disables writes on the socket endpoint.
    WriteDisabled = ZX_SOCKET_DISPOSITION_WRITE_DISABLED,
    /// Enables writes on the socket endpoint.
    WriteEnabled = ZX_SOCKET_DISPOSITION_WRITE_ENABLED,
}

impl TryFrom<u32> for Disposition {
    type Error = Status;

    /// Converts a raw uint32_t disposition into a `Disposition` enum variant.
    fn try_from(disposition: u32) -> Result<Self, Status> {
        match disposition {
            0 => Ok(Self::None),
            ZX_SOCKET_DISPOSITION_WRITE_DISABLED => Ok(Self::WriteDisabled),
            ZX_SOCKET_DISPOSITION_WRITE_ENABLED => Ok(Self::WriteEnabled),
            _ => Err(Status::INVALID_ARGS),
        }
    }
}

/// Signal masks generated by a [`Disposition`].
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
struct SignalMasks {
    /// Signals to clear on the target endpoint.
    clear: zx_signals_t,
    /// Signals to set on the target endpoint.
    set: zx_signals_t,
    /// Signals to clear on the peer endpoint.
    peer_clear: zx_signals_t,
    /// Signals to set on the peer endpoint.
    peer_set: zx_signals_t,
}

impl Disposition {
    /// Returns the signal masks corresponding to this disposition.
    fn signal_masks(self) -> SignalMasks {
        match self {
            Self::WriteDisabled => SignalMasks {
                clear: ZX_SOCKET_WRITABLE,
                set: ZX_SOCKET_WRITE_DISABLED,
                peer_clear: 0,
                peer_set: ZX_SOCKET_PEER_WRITE_DISABLED,
            },
            Self::WriteEnabled => SignalMasks {
                clear: ZX_SOCKET_WRITE_DISABLED,
                set: ZX_SOCKET_WRITABLE,
                peer_clear: ZX_SOCKET_PEER_WRITE_DISABLED,
                peer_set: 0,
            },
            Self::None => SignalMasks::default(),
        }
    }
}

/// Specifies the type of read operation (consume or peek).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReadType {
    /// Consumes read data from the socket.
    Consume,
    /// Peeks data without consuming it from the socket buffer.
    Peek,
}

/// Synchronization and state layout for `SocketDispatcher`.
///
/// The shared `mu` lock protects all state members below.
#[guarded]
#[pin_data(PinnedDrop)]
#[repr(C)]
pub struct SocketDispatcherState {
    canary: Canary<{ fbl::magic(b"SOCK") }>,

    #[pin]
    pub peered: PeeredState<SocketDispatcher>,

    flags: u32,
    #[guarded_by(mu)]
    read_disabled: bool,

    #[guarded_by(mu)]
    read_threshold: usize,
    #[guarded_by(mu)]
    write_threshold: usize,

    #[guarded_by(mu)]
    #[pin]
    data: MBufChain,

    #[mutex(PeerHolderMuClass<SocketDispatcher>)]
    pub mu: KMutex<PhantomMutex>,
}

impl SocketDispatcherState {
    /// Initializes `SocketDispatcherState` with peer holder and creation flags.
    pub fn init(
        holder: RefPtr<PeerHolder<SocketDispatcher>>,
        flags: u32,
    ) -> impl PinInit<Self, Infallible> {
        DISPATCHER_SOCKET_CREATE_COUNT.add(1);
        pin_init!(Self {
            canary: Canary::new(),
            peered <- PeeredState::init(holder),
            flags,
            read_disabled: false.into(),
            read_threshold: 0.into(),
            write_threshold: 0.into(),
            data <- ksync::kcell_init(MBufChain::new()),
            mu: KMutex::new(PhantomMutex),
        })
    }
}

#[pinned_drop]
impl PinnedDrop for SocketDispatcherState {
    fn drop(self: Pin<&mut Self>) {
        DISPATCHER_SOCKET_DESTROY_COUNT.add(1);
    }
}

impl_peered_dispatcher_facade_with_state!(
    pub struct SocketDispatcher,
    SocketDispatcherState,
    ZX_OBJ_TYPE_SOCKET,
    kSocketDispatcherStateOffset,
    allowed_signals: ALLOWED_SIGNALS,
);

impl SocketDispatcher {
    /// Returns default rights for a SocketDispatcher.
    pub fn default_rights() -> zx_rights_t {
        DEFAULT_RIGHTS
    }

    /// Creates a new SocketDispatcher pair and returns their kernel handles and rights.
    pub fn create(
        flags: u32,
    ) -> Result<(KernelHandle<Self>, KernelHandle<Self>, zx_rights_t), Status> {
        ltrace_entry!();

        if (flags & !ZX_SOCKET_CREATE_MASK) != 0 {
            return Err(Status::INVALID_ARGS);
        }

        let holder0 = PeerHolder::<Self>::create().map_err(|_| Status::NO_MEMORY)?;
        let holder1 = holder0.clone();

        let create_single =
            |holder: RefPtr<PeerHolder<Self>>| -> Result<KernelHandle<Self>, Status> {
                // SAFETY: `RefPtr::into_raw(holder)` converts the valid reference into a raw
                // pointer, and `cpp_socket_dispatcher_create` initializes `handle` on success.
                unsafe {
                    KernelHandle::create(|out| {
                        cpp_socket_dispatcher_create(RefPtr::into_raw(holder) as *mut _, flags, out)
                    })
                }
            };

        let handle0 = create_single(holder0)?;
        let handle1 = create_single(holder1)?;

        handle0.dispatcher().init_peer(handle1.dispatcher().clone());
        handle1.dispatcher().init_peer(handle0.dispatcher().clone());

        Ok((handle0, handle1, DEFAULT_RIGHTS))
    }

    fn on_zero_handles_locked(&self, _token: &ksync::LockToken<'_, PeerHolderMuClass<Self>>) {
        self.state().canary.assert();
    }

    fn on_peer_zero_handles_locked(&self, token: &ksync::LockToken<'_, PeerHolderMuClass<Self>>) {
        self.state().canary.assert();
        self.update_state_locked(token, ZX_SOCKET_WRITABLE, ZX_SOCKET_PEER_CLOSED);
    }

    fn update_read_status(
        &self,
        token: &mut ksync::LockToken<'_, PeerHolderMuClass<SocketDispatcher>>,
        disposition_peer: Disposition,
    ) {
        match disposition_peer {
            Disposition::WriteDisabled => {
                *self.state().guard_mu_mut(token).read_disabled_mut() = true;
            }
            Disposition::WriteEnabled => {
                *self.state().guard_mu_mut(token).read_disabled_mut() = false;
            }
            Disposition::None => {}
        }
    }

    fn is_disposition_state_valid(
        &self,
        token: &mut ksync::LockToken<'_, PeerHolderMuClass<SocketDispatcher>>,
        disposition_peer: Disposition,
    ) -> bool {
        // All the data written by an endpoint must be read by the other endpoint before writes can
        // be re-enabled, as per /docs/reference/syscalls/socket_set_disposition.md.
        !(disposition_peer == Disposition::WriteEnabled
            && !self.state().guard_mu(token).data().is_empty()
            && (self.signals_state_locked(token) & ZX_SOCKET_PEER_WRITE_DISABLED) != 0)
    }

    /// Sets the disposition of socket endpoints (write enabled / write disabled).
    pub fn set_disposition(
        &self,
        disposition: Disposition,
        disposition_peer: Disposition,
    ) -> Result<(), Status> {
        self.state().canary.assert();
        ltrace_entry!();

        if disposition == Disposition::None && disposition_peer == Disposition::None {
            // Nothing to do, return early.
            return Ok(());
        }

        let self_masks = disposition.signal_masks();

        ksync::lock!(let mut guard = self.state().peered.lock());
        let peer = guard.peer().as_ref().cloned();
        let token = guard.as_mut().token_mut();

        if !self.is_disposition_state_valid(token, disposition_peer) {
            return Err(Status::BAD_STATE);
        }

        let mut clear_mask = self_masks.clear;
        let mut set_mask = self_masks.set;

        if let Some(peer) = peer.as_ref() {
            if !peer.is_disposition_state_valid(token, disposition) {
                return Err(Status::BAD_STATE);
            }
            let peer_masks = disposition_peer.signal_masks();
            let clear_mask_peer = self_masks.peer_clear | peer_masks.clear;
            let set_mask_peer = self_masks.peer_set | peer_masks.set;
            clear_mask |= peer_masks.peer_clear;
            set_mask |= peer_masks.peer_set;

            peer.update_read_status(token, disposition);
            peer.update_state_locked(token, clear_mask_peer, set_mask_peer);
        }

        self.update_read_status(token, disposition_peer);
        self.update_state_locked(token, clear_mask, set_mask);

        Ok(())
    }

    /// Writes data from the buffer pointer into the peer socket endpoint.
    pub fn write(&self, src: UserInPtr<core::ffi::c_char>, len: usize) -> Result<usize, Status> {
        self.state().canary.assert();
        ltrace_entry!();

        ksync::lock!(let mut guard = self.state().peered.lock());
        let peer = guard.peer().clone().ok_or(Status::PEER_CLOSED)?;

        if (self.signals_state_locked(guard.token()) & ZX_SOCKET_WRITE_DISABLED) != 0 {
            return Err(Status::BAD_STATE);
        }
        if len == 0 {
            return Ok(0);
        }
        if u32::try_from(len).is_err() {
            return Err(Status::INVALID_ARGS);
        }

        peer.write_self_locked(self, guard.as_mut(), src, len)
    }

    fn write_self_locked(
        &self,
        peer: &Self,
        mut guard: Pin<&mut PeeredStateMuGuard<'_, SocketDispatcher, RawCriticalMutex>>,
        src: UserInPtr<core::ffi::c_char>,
        len: usize,
    ) -> Result<usize, Status> {
        self.state().canary.assert();

        let was_empty = {
            let state = self.state().guard_mu(guard.token());
            if state.data().is_full() {
                return Err(Status::SHOULD_WAIT);
            }
            state.data().is_empty()
        };

        // TODO(https://fxbug.dev/42182048): Perform user copying while holding the dispatcher lock
        // is generally not allowed, but is exempted here while a fix for sockets is developed.
        // Performing the MBufChain operations (which do the actual user copy) with tracking
        // disabled will allow the user copy to go through, with side effect of reducing the
        // effectiveness of any other lockdep detections that might involve this lock for the
        // duration of the operation.
        let (status, written) = guard.as_mut().call_untracked(|token| {
            let is_datagram = (self.state().flags & ZX_SOCKET_DATAGRAM) != 0;
            let mut state = self.state().guard_mu_mut(token);
            if is_datagram {
                state.data_mut().write_datagram(src, len)
            } else {
                state.data_mut().write_stream(src, len)
            }
        });

        // Regardless of the status, data may have been added, and so we need to update the signals.
        let mut peer_clear: zx_signals_t = 0;
        let state = self.state().guard_mu(guard.token());

        if written > 0 {
            let mut set: zx_signals_t = 0;
            if was_empty {
                set |= ZX_SOCKET_READABLE;
            }
            // Assert signal if we go above the read threshold
            let read_threshold = *state.read_threshold();
            let size = state.data().stream_size();
            if read_threshold > 0 && size >= read_threshold {
                set |= ZX_SOCKET_READ_THRESHOLD;
            }
            if set != 0 {
                self.update_state_locked(guard.token(), 0, set);
            }

            let peer_write_threshold = *peer.state().guard_mu(guard.token()).write_threshold();
            let free_space = MBufChain::MAX_SIZE - size;
            // If free space falls below threshold, de-signal
            if peer_write_threshold > 0 && free_space < peer_write_threshold {
                peer_clear |= ZX_SOCKET_WRITE_THRESHOLD;
            }
        }

        if state.data().is_full() {
            peer_clear |= ZX_SOCKET_WRITABLE;
        }
        if peer_clear != 0 {
            peer.update_state_locked(guard.token(), peer_clear, 0);
        }

        status?;
        Ok(written)
    }

    /// Reads data from this socket endpoint into the destination buffer pointer.
    pub fn read(
        &self,
        read_type: ReadType,
        dst: UserOutPtr<core::ffi::c_char>,
        len: usize,
    ) -> Result<usize, Status> {
        self.state().canary.assert();
        ltrace_entry!();

        ksync::lock!(let mut guard = self.state().peered.lock());

        if u32::try_from(len).is_err() {
            return Err(Status::INVALID_ARGS);
        }

        let was_full = {
            let state = self.state().guard_mu(guard.token());
            if state.data().is_empty() {
                if guard.peer().is_none() {
                    return Err(Status::PEER_CLOSED);
                }
                // If reading is disabled on our end and we're empty, we'll never become readable
                // again. Return a different error to let the caller know.
                if *state.read_disabled() {
                    return Err(Status::BAD_STATE);
                }
                return Err(Status::SHOULD_WAIT);
            }
            state.data().is_full()
        };

        // TODO(https://fxbug.dev/42182048): See comment in write_self_locked on why we use
        // call_untracked.
        let (status, actual) = guard.as_mut().call_untracked(|token| {
            let is_datagram = (self.state().flags & ZX_SOCKET_DATAGRAM) != 0;
            let mut state = self.state().guard_mu_mut(token);
            match (read_type, is_datagram) {
                (ReadType::Peek, true) => state.data().peek_datagram(dst, len),
                (ReadType::Peek, false) => state.data().peek_stream(dst, len),
                (ReadType::Consume, true) => state.data_mut().read_datagram(dst, len),
                (ReadType::Consume, false) => state.data_mut().read_stream(dst, len),
            }
        });

        if read_type == ReadType::Peek {
            status?;
            return Ok(actual);
        }

        // Regardless of the status, data may have been consumed, and so we need to update the
        // signals.
        let mut clear: zx_signals_t = 0;
        let state = self.state().guard_mu(guard.token());

        // Deassert signal if we fell below the read threshold
        let read_threshold = *state.read_threshold();
        let size = state.data().stream_size();
        if read_threshold > 0 && size < read_threshold {
            clear |= ZX_SOCKET_READ_THRESHOLD;
        }
        if state.data().is_empty() {
            clear |= ZX_SOCKET_READABLE;
        }
        if clear != 0 {
            self.update_state_locked(guard.token(), clear, 0);
        }

        if let Some(peer) = guard.peer().as_ref() {
            let peer_write_threshold = *peer.state().guard_mu(guard.token()).write_threshold();
            let free_space = MBufChain::MAX_SIZE - size;
            let mut peer_set: zx_signals_t = 0;
            // Assert (write threshold) signal if space available is above threshold.
            if peer_write_threshold > 0 && free_space >= peer_write_threshold {
                peer_set |= ZX_SOCKET_WRITE_THRESHOLD;
            }
            if was_full && actual > 0 {
                peer_set |= ZX_SOCKET_WRITABLE;
            }
            if peer_set != 0 {
                peer.update_state_locked(guard.token(), 0, peer_set);
            }
        }

        status?;
        Ok(actual)
    }

    /// Returns socket buffer info for this endpoint and its peer.
    pub fn get_info(&self) -> zx_info_socket_t {
        self.state().canary.assert();
        ksync::lock!(let mut guard = self.state().peered.lock());
        let peer = guard.peer().as_ref().cloned();
        let token = guard.as_mut().token_mut();
        let is_datagram = (self.state().flags & ZX_SOCKET_DATAGRAM) != 0;

        let guard_mu = self.state().guard_mu(token);
        let data = guard_mu.data();
        let mut info = zx_info_socket_t {
            options: self.state().flags,
            rx_buf_max: MBufChain::MAX_SIZE,
            rx_buf_size: data.stream_size(),
            rx_buf_available: if is_datagram { data.datagram_size() } else { data.stream_size() },
            tx_buf_max: 0,
            tx_buf_size: 0,
            ..Default::default()
        };

        if let Some(peer) = peer {
            // Peer shares the same lock.
            let peer_guard_mu = peer.state().guard_mu(token);
            info.tx_buf_max = MBufChain::MAX_SIZE;
            info.tx_buf_size = peer_guard_mu.data().stream_size();
        }
        info
    }

    /// Returns the read threshold value.
    pub fn get_read_threshold(&self) -> usize {
        self.state().canary.assert();
        ksync::lock!(let guard = self.state().peered.lock());
        *self.state().guard_mu(guard.token()).read_threshold()
    }

    /// Returns the write threshold value.
    pub fn get_write_threshold(&self) -> usize {
        self.state().canary.assert();
        ksync::lock!(let guard = self.state().peered.lock());
        *self.state().guard_mu(guard.token()).write_threshold()
    }

    /// Sets the read threshold value.
    pub fn set_read_threshold(&self, value: usize) -> Result<(), Status> {
        self.state().canary.assert();
        ksync::lock!(let mut guard = self.state().peered.lock());

        if value > MBufChain::MAX_SIZE {
            return Err(Status::INVALID_ARGS);
        }

        let token = guard.as_mut().token_mut();
        let mut state = self.state().guard_mu_mut(token);
        let size = state.data().stream_size();
        *state.read_threshold_mut() = value;

        // Setting 0 disables thresholding. Deassert signal unconditionally.
        // Assert signal if we have queued data above the read threshold.
        // De-assert signal if we upped threshold and queued data drops below.
        let (clear, set) = if value > 0 && size >= value {
            (0, ZX_SOCKET_READ_THRESHOLD)
        } else {
            (ZX_SOCKET_READ_THRESHOLD, 0)
        };
        self.update_state_locked(token, clear, set);
        Ok(())
    }

    /// Sets the write threshold value.
    pub fn set_write_threshold(&self, value: usize) -> Result<(), Status> {
        self.state().canary.assert();
        ksync::lock!(let mut guard = self.state().peered.lock());

        if value > MBufChain::MAX_SIZE {
            return Err(Status::INVALID_ARGS);
        }

        let peer_available = {
            let peer = guard.peer().as_ref().ok_or(Status::PEER_CLOSED)?;
            MBufChain::MAX_SIZE - peer.state().guard_mu(guard.token()).data().stream_size()
        };
        let token = guard.as_mut().token_mut();
        *self.state().guard_mu_mut(token).write_threshold_mut() = value;

        // Setting 0 disables thresholding. Deassert signal unconditionally.
        // Assert signal if we have available space above the write threshold.
        // De-assert signal if we upped threshold and available space drops below.
        let (clear, set) = if value > 0 && peer_available >= value {
            (0, ZX_SOCKET_WRITE_THRESHOLD)
        } else {
            (ZX_SOCKET_WRITE_THRESHOLD, 0)
        };
        self.update_state_locked(token, clear, set);
        Ok(())
    }
}

/// In-tree kernel unit tests for `SocketDispatcher`.
#[cfg(ktest)]
#[unittest::suite(name = "socket_dispatcher_rust_tests")]
mod tests {
    use super::{DEFAULT_RIGHTS, Disposition, ReadType, SocketDispatcher};
    use crate::object::dispatcher::DispatcherOps;
    use crate::user_copy::{UserInPtr, UserOutPtr};
    use crate::user_memory::UserMemory;
    use core::ffi::c_char;
    use core::mem::MaybeUninit;
    use kalloc::Box;
    use unittest::{expect_eq, expect_ok, expect_true};
    use zx_status::Status;
    use zx_types::{
        ZX_SOCKET_DATAGRAM, ZX_SOCKET_DISPOSITION_WRITE_DISABLED,
        ZX_SOCKET_DISPOSITION_WRITE_ENABLED, ZX_SOCKET_PEER_CLOSED, ZX_SOCKET_PEER_WRITE_DISABLED,
        ZX_SOCKET_READ_THRESHOLD, ZX_SOCKET_READABLE, ZX_SOCKET_WRITABLE, ZX_SOCKET_WRITE_DISABLED,
        ZX_SOCKET_WRITE_THRESHOLD,
    };

    fn make_user_in_pattern(
        size: usize,
        pattern: impl Fn(usize) -> u8,
    ) -> Option<(UserMemory, UserInPtr<c_char>)> {
        let alloc_size = if size == 0 { 1 } else { size };
        let mem = UserMemory::create(alloc_size)?;
        mem.commit_and_map(0..alloc_size).ok()?;
        let mut chunk = [0u8; 512];
        let mut offset = 0;
        while offset < size {
            let to_write = core::cmp::min(chunk.len(), size - offset);
            for (i, b) in chunk[..to_write].iter_mut().enumerate() {
                *b = pattern(offset + i);
            }
            mem.vmo_write(&chunk[..to_write], offset as u64).ok()?;
            offset += to_write;
        }
        let ptr = UserInPtr::new(mem.base() as *const c_char);
        Some((mem, ptr))
    }

    fn make_user_in(data: &[u8]) -> Option<(UserMemory, UserInPtr<c_char>)> {
        make_user_in_pattern(data.len(), |i| data[i])
    }

    fn make_user_out(size: usize) -> Option<(UserMemory, UserOutPtr<c_char>)> {
        let alloc_size = if size == 0 { 1 } else { size };
        let mem = UserMemory::create(alloc_size)?;
        mem.commit_and_map(0..alloc_size).ok()?;
        let ptr = UserOutPtr::new(mem.base() as *mut c_char);
        Some((mem, ptr))
    }

    fn read_user_mem(mem: &UserMemory, size: usize) -> Option<Box<[u8]>> {
        let mut buf = Box::try_new_uninit_slice(size).ok()?;
        mem.vmo_read(&mut buf, 0).ok()?;
        // SAFETY: We initialize the buffer immediately by reading `size` bytes from the VMO.
        Some(unsafe { buf.assume_init() })
    }

    /// Allocate/destroy many sockets. Ad hoc resource leak check.
    #[test]
    fn test_create_destroy_many_sockets() {
        const MANY: usize = 100_000;
        for flags in [0, ZX_SOCKET_DATAGRAM] {
            for _ in 0..MANY {
                let (h0, h1, _rights) =
                    SocketDispatcher::create(flags).expect("failed to create socket");
                drop(h0);
                drop(h1);
            }
        }
    }

    /// Stream socket write/read test matching C++ TestCreateWriteReadClose.
    #[test]
    fn test_create_write_read_close() {
        const SIZE: usize = 3357;
        let (h0, h1, rights) = SocketDispatcher::create(0).expect("failed to create stream socket");
        expect_eq!(rights, DEFAULT_RIGHTS);
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        let info0 = d0.get_info();
        expect_eq!(info0.rx_buf_available, 0); // No bytes written yet.

        // Write a test pattern, read it back one byte at a time.
        let (_in_mem, in_ptr) = make_user_in_pattern(SIZE, |i| (i & 0xff) as u8).unwrap();
        let written = d0.write(in_ptr, SIZE).expect("failed to write");
        expect_eq!(written, SIZE);

        // Expect to not be able to read on the dispatcher side you just wrote to
        let info0 = d0.get_info();
        expect_eq!(info0.rx_buf_available, 0);
        // Expect to be able to read from the paired dispatcher.
        let info1 = d1.get_info();
        expect_eq!(info1.rx_buf_available, SIZE);

        // Read out data from the peer byte-at-a-time; this is a stream socket, allowing that.
        let (out_mem, out_ptr) = make_user_out(1).unwrap();
        let mut read_buffer = [0u8; SIZE];
        for (i, slot) in read_buffer.iter_mut().enumerate() {
            let bytes_read = d1.read(ReadType::Consume, out_ptr, 1).expect("failed to read");
            expect_eq!(bytes_read, 1);
            let mut byte_buf = [MaybeUninit::<u8>::uninit(); 1];
            let read = out_mem.vmo_read(&mut byte_buf, 0).expect("vmo read");
            *slot = read[0];
            // Expect consuming 1-byte reads to reduce rx_buf_available.
            let info1 = d1.get_info();
            expect_eq!(info1.rx_buf_available, SIZE - (i + 1));
        }
        for (i, &val) in read_buffer.iter().enumerate() {
            expect_eq!(val, (i & 0xff) as u8);
        }

        // Test that shutting down a socket for writes still allows reads from the paired
        // dispatcher.
        let byte = [0u8; 1];
        let (_in_mem, in_ptr) = make_user_in(&byte).expect("make user in");
        let written = d0.write(in_ptr, 1).expect("failed to write");
        expect_eq!(written, 1);

        let res = d0.set_disposition(Disposition::WriteDisabled, Disposition::None);
        expect_ok!(res);

        let res = d0.write(in_ptr, 1);
        // |written| is not updated if Write() fails.
        expect_true!(res == Err(Status::BAD_STATE));

        let info1 = d1.get_info();
        // Not 2 - the second write must have failed.
        expect_eq!(info1.rx_buf_available, 1);
    }

    /// Proves that Disposition match is exhaustive in Rust.
    #[test]
    fn test_disposition_switch_must_be_exhaustive() {
        let check_exhaustive = |disposition: Disposition| match disposition {
            Disposition::None => true,
            Disposition::WriteDisabled => true,
            Disposition::WriteEnabled => true,
            // This proves that exhaustive checking is done by the match - otherwise we would get
            // a compilation error because we would not return a value in all code paths.
        };
        expect_true!(check_exhaustive(Disposition::None));
        expect_true!(check_exhaustive(Disposition::WriteDisabled));
        expect_true!(check_exhaustive(Disposition::WriteEnabled));
    }

    /// Tests Disposition::try_from conversions and error validation.
    #[test]
    fn test_disposition_try_from() {
        expect_true!(Disposition::try_from(0) == Ok(Disposition::None));
        expect_true!(
            Disposition::try_from(ZX_SOCKET_DISPOSITION_WRITE_DISABLED)
                == Ok(Disposition::WriteDisabled)
        );
        expect_true!(
            Disposition::try_from(ZX_SOCKET_DISPOSITION_WRITE_ENABLED)
                == Ok(Disposition::WriteEnabled)
        );
        expect_true!(Disposition::try_from(0xffff) == Err(Status::INVALID_ARGS));
    }

    /// Datagram socket write, peek, truncate on short read, and full read.
    #[test]
    fn test_datagram_write_read_peek_truncate() {
        let (h0, h1, rights) =
            SocketDispatcher::create(ZX_SOCKET_DATAGRAM).expect("failed to create datagram socket");
        expect_eq!(rights, DEFAULT_RIGHTS);
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        let msg1 = b"12345";
        let msg2 = b"abcdef";

        let (_m1_mem, m1_ptr) = make_user_in(msg1).unwrap();
        let (_m2_mem, m2_ptr) = make_user_in(msg2).unwrap();
        let (_m3_mem, m3_ptr) = make_user_in_pattern(100, |i| (i & 0xff) as u8).unwrap();

        // Write three distinct datagrams into d1 via d0.
        let w1 = d0.write(m1_ptr, msg1.len()).expect("write datagram 1");
        expect_eq!(w1, msg1.len());

        let w2 = d0.write(m2_ptr, msg2.len()).expect("write datagram 2");
        expect_eq!(w2, msg2.len());

        let w3 = d0.write(m3_ptr, 100).expect("write datagram 3");
        expect_eq!(w3, 100);

        // Available bytes for datagram sockets should report the head datagram size.
        let info = d1.get_info();
        expect_eq!(info.rx_buf_available, msg1.len());
        expect_eq!(info.rx_buf_size, msg1.len() + msg2.len() + 100);

        // Short peek on the first datagram: only peek part of msg1.
        let (peek_mem, peek_ptr) = make_user_out(4).unwrap();
        let npeek = d1.read(ReadType::Peek, peek_ptr, 4).expect("peek short");
        expect_eq!(npeek, 4);
        let b = read_user_mem(&peek_mem, 4).unwrap();
        expect_true!(*b == msg1[..4]);

        // Buffer info must remain unchanged after peek.
        let info = d1.get_info();
        expect_eq!(info.rx_buf_available, msg1.len());
        expect_eq!(info.rx_buf_size, msg1.len() + msg2.len() + 100);

        // Full peek with larger buffer should only return the first datagram.
        let (peek_large_mem, peek_large_ptr) = make_user_out(16).unwrap();
        let npeek_large = d1.read(ReadType::Peek, peek_large_ptr, 16).expect("peek large");
        expect_eq!(npeek_large, msg1.len());
        let b = read_user_mem(&peek_large_mem, msg1.len()).unwrap();
        expect_true!(*b == *msg1);

        // Short consuming read should truncate and discard the rest of msg1.
        let (read_trunc_mem, read_trunc_ptr) = make_user_out(3).unwrap();
        let nread_trunc = d1.read(ReadType::Consume, read_trunc_ptr, 3).expect("read truncate");
        expect_eq!(nread_trunc, 3);
        let b = read_user_mem(&read_trunc_mem, 3).unwrap();
        expect_true!(*b == msg1[..3]);

        // Next available datagram is msg2.
        let info = d1.get_info();
        expect_eq!(info.rx_buf_available, msg2.len());
        expect_eq!(info.rx_buf_size, msg2.len() + 100);

        // Consume read msg2 completely.
        let (read_msg2_mem, read_msg2_ptr) = make_user_out(6).unwrap();
        let nread2 = d1.read(ReadType::Consume, read_msg2_ptr, 6).expect("read msg2");
        expect_eq!(nread2, msg2.len());
        let b = read_user_mem(&read_msg2_mem, 6).unwrap();
        expect_true!(&*b == msg2);

        // Consume read msg3 with an oversized buffer.
        let (read_msg3_mem, read_msg3_ptr) = make_user_out(128).unwrap();
        let nread3 = d1.read(ReadType::Consume, read_msg3_ptr, 128).expect("read msg3");
        expect_eq!(nread3, 100);
        let b = read_user_mem(&read_msg3_mem, 100).unwrap();
        for (i, &val) in b.iter().enumerate() {
            expect_eq!(val, (i & 0xff) as u8);
        }

        // Socket is now empty; reading should return SHOULD_WAIT.
        let (_empty_mem, empty_ptr) = make_user_out(1).unwrap();
        let res = d1.read(ReadType::Consume, empty_ptr, 1);
        expect_true!(res == Err(Status::SHOULD_WAIT));

        let info = d1.get_info();
        expect_eq!(info.rx_buf_available, 0);
        expect_eq!(info.rx_buf_size, 0);
    }

    /// Tests read and write threshold setting, retrieval, and signal transitions.
    #[test]
    fn test_read_write_thresholds_and_signals() {
        let (h0, h1, _) = SocketDispatcher::create(0).expect("failed to create stream socket");
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        // Initial threshold values default to 0 (disabled).
        expect_eq!(d0.get_read_threshold(), 0);
        expect_eq!(d0.get_write_threshold(), 0);

        let info0 = d0.get_info();
        // Setting threshold beyond capacity must fail with INVALID_ARGS.
        expect_true!(d0.set_read_threshold(info0.rx_buf_max + 1) == Err(Status::INVALID_ARGS));
        expect_true!(d0.set_write_threshold(info0.rx_buf_max + 1) == Err(Status::INVALID_ARGS));

        // Setting a valid read threshold of 50 bytes.
        expect_ok!(d0.set_read_threshold(50));
        expect_eq!(d0.get_read_threshold(), 50);

        // Signal should not be active since 0 < 50.
        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_eq!(sigs0 & ZX_SOCKET_READ_THRESHOLD, 0);
        }

        // Write 30 bytes into d0 from d1 (below read threshold).
        let (_b30_mem, b30_ptr) = make_user_in(&[0xaa_u8; 30]).unwrap();
        let w = d1.write(b30_ptr, 30).expect("write 30");
        expect_eq!(w, 30);

        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_true!((sigs0 & ZX_SOCKET_READABLE) != 0);
            expect_eq!(sigs0 & ZX_SOCKET_READ_THRESHOLD, 0);
        }

        // Write 25 more bytes (total 55 >= 50 threshold).
        let (_b25_mem, b25_ptr) = make_user_in(&[0xbb_u8; 25]).unwrap();
        let w = d1.write(b25_ptr, 25).expect("write 25");
        expect_eq!(w, 25);

        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_true!((sigs0 & ZX_SOCKET_READ_THRESHOLD) != 0);
        }

        // Read 10 bytes from d0 (leaving 45 bytes, falling below the 50 threshold).
        let (_r10_mem, r10_ptr) = make_user_out(10).unwrap();
        let nread = d0.read(ReadType::Consume, r10_ptr, 10).expect("read 10");
        expect_eq!(nread, 10);

        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_eq!(sigs0 & ZX_SOCKET_READ_THRESHOLD, 0);
            expect_true!((sigs0 & ZX_SOCKET_READABLE) != 0);
        }

        // Setting read threshold to 0 disables it and clears signal unconditionally.
        expect_ok!(d0.set_read_threshold(0));
        expect_eq!(d0.get_read_threshold(), 0);

        // Test write threshold on d1 (which tracks available space in d0).
        expect_ok!(d1.set_write_threshold(100));
        expect_eq!(d1.get_write_threshold(), 100);

        {
            ksync::lock!(let guard1 = d1.state().peered.lock());
            let sigs1 = d1.signals_state_locked(guard1.token());
            // Since d0 has ample space (> 100 bytes available), write threshold signal is asserted.
            expect_true!((sigs1 & ZX_SOCKET_WRITE_THRESHOLD) != 0);
        }

        // Fill d0 almost to capacity so available space drops below 100 bytes.
        let to_fill = info0.rx_buf_max - 50;
        let (_fill_mem, fill_ptr) = make_user_in_pattern(to_fill, |_| 0xcc).unwrap();
        let w = d1.write(fill_ptr, to_fill).expect("write fill");
        expect_eq!(w, to_fill);

        {
            ksync::lock!(let guard1 = d1.state().peered.lock());
            let sigs1 = d1.signals_state_locked(guard1.token());
            // Available space in d0 is now 50 (< 100 write threshold), so signal should be de-asserted.
            expect_eq!(sigs1 & ZX_SOCKET_WRITE_THRESHOLD, 0);
            expect_true!((sigs1 & ZX_SOCKET_WRITABLE) != 0);
        }

        // Disabling write threshold by setting it to 0.
        expect_ok!(d1.set_write_threshold(0));
        expect_eq!(d1.get_write_threshold(), 0);

        {
            ksync::lock!(let guard1 = d1.state().peered.lock());
            let sigs1 = d1.signals_state_locked(guard1.token());
            expect_eq!(sigs1 & ZX_SOCKET_WRITE_THRESHOLD, 0);
        }
    }

    /// Tests socket operations and signal handling when the peer endpoint is closed.
    #[test]
    fn test_closed_peer_operations() {
        let (h0, h1, _) = SocketDispatcher::create(0).expect("failed to create stream socket");
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        // Write some data from d0 to d1 before closing d0.
        let (_test_mem, test_ptr) = make_user_in(&[42u8; 16]).unwrap();
        let written = d0.write(test_ptr, 16).expect("write test data");
        expect_eq!(written, 16);

        // Close peer endpoint d0.
        drop(h0);

        // Peer closing asserts ZX_SOCKET_PEER_CLOSED and de-asserts ZX_SOCKET_WRITABLE on d1.
        {
            ksync::lock!(let guard1 = d1.state().peered.lock());
            let sigs1 = d1.signals_state_locked(guard1.token());
            expect_true!((sigs1 & ZX_SOCKET_PEER_CLOSED) != 0);
            expect_eq!(sigs1 & ZX_SOCKET_WRITABLE, 0);
        }

        // Writing to d1 when peer is closed must return PEER_CLOSED (even for 0 bytes).
        let res_w = d1.write(test_ptr, 16);
        expect_true!(res_w == Err(Status::PEER_CLOSED));
        let res_w0 = d1.write(test_ptr, 0);
        expect_true!(res_w0 == Err(Status::PEER_CLOSED));

        // Setting write threshold when peer is closed must return PEER_CLOSED.
        let res_wt = d1.set_write_threshold(50);
        expect_true!(res_wt == Err(Status::PEER_CLOSED));

        // Getting write threshold still succeeds.
        expect_eq!(d1.get_write_threshold(), 0);

        // Setting read threshold still works on closed peer.
        expect_ok!(d1.set_read_threshold(10));
        expect_eq!(d1.get_read_threshold(), 10);

        // Getting socket info on closed peer returns rx info and zeroed tx capacity.
        let info = d1.get_info();
        expect_eq!(info.rx_buf_size, 16);
        expect_eq!(info.tx_buf_max, 0);
        expect_eq!(info.tx_buf_size, 0);

        // Buffered data can still be read out completely.
        let (r_mem, r_ptr) = make_user_out(16).unwrap();
        let nread = d1.read(ReadType::Consume, r_ptr, 16).expect("read remaining data");
        expect_eq!(nread, 16);
        let b = read_user_mem(&r_mem, 16).unwrap();
        expect_true!(*b == [42u8; 16]);

        // Once empty and peer is closed, reading returns PEER_CLOSED.
        let res_r = d1.read(ReadType::Consume, r_ptr, 16);
        expect_true!(res_r == Err(Status::PEER_CLOSED));

        // Setting disposition on closed peer: no-op disposition succeeds.
        expect_ok!(d1.set_disposition(Disposition::None, Disposition::None));
        expect_ok!(d1.set_disposition(Disposition::WriteDisabled, Disposition::None));
    }

    /// Tests creating socket with invalid flags returns INVALID_ARGS.
    #[test]
    fn test_invalid_flags_creation() {
        let res = SocketDispatcher::create(0xffff_0000);
        expect_true!(matches!(res, Err(Status::INVALID_ARGS)));
    }

    /// Tests filling stream socket to capacity, WRITABLE signal deassertion, and recovery on read.
    #[test]
    fn test_stream_socket_full_and_writable_signals() {
        let (h0, h1, _) = SocketDispatcher::create(0).expect("failed to create stream socket");
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        let info = d0.get_info();
        let capacity = info.rx_buf_max;
        expect_true!(capacity > 0);

        // Fill socket almost completely.
        const CHUNK: usize = 32 * 1024;
        let (_in_mem, in_ptr) = make_user_in_pattern(CHUNK, |i| (i & 0xff) as u8).unwrap();
        let mut total_written = 0usize;

        while total_written + CHUNK <= capacity {
            let w = d0.write(in_ptr, CHUNK).expect("write chunk");
            expect_eq!(w, CHUNK);
            total_written += w;
        }

        // Top off with exact remaining bytes to reach full capacity.
        let rem = capacity - total_written;
        if rem > 0 {
            let (_rem_mem, rem_ptr) = make_user_in_pattern(rem, |_| 0xaa).unwrap();
            let w = d0.write(rem_ptr, rem).expect("write remaining");
            expect_eq!(w, rem);
            total_written += w;
        }

        expect_eq!(total_written, capacity);
        let info1 = d1.get_info();
        expect_eq!(info1.rx_buf_size, capacity);

        // Socket is now full; writer (d0) should have ZX_SOCKET_WRITABLE deasserted.
        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_eq!(sigs0 & ZX_SOCKET_WRITABLE, 0);
        }

        // Writing when full returns SHOULD_WAIT.
        let res_overflow = d0.write(in_ptr, 1);
        expect_true!(res_overflow == Err(Status::SHOULD_WAIT));

        // Reading even 1 byte from receiver (d1) should immediately re-assert ZX_SOCKET_WRITABLE.
        let (_out_mem, out_ptr) = make_user_out(1).unwrap();
        let nread = d1.read(ReadType::Consume, out_ptr, 1).expect("read 1 byte");
        expect_eq!(nread, 1);

        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_true!((sigs0 & ZX_SOCKET_WRITABLE) != 0);
        }

        // Test short write: attempt to write 100 bytes when only 1 byte is available.
        let (_short_mem, short_ptr) = make_user_in_pattern(100, |_| 0xbb).unwrap();
        let w_short = d0.write(short_ptr, 100).expect("short write");
        expect_eq!(w_short, 1);

        // Socket is full again.
        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_eq!(sigs0 & ZX_SOCKET_WRITABLE, 0);
        }
    }

    /// Tests disposition state transitions, write-disabled errors, and re-enabling restrictions.
    #[test]
    fn test_disposition_state_transitions() {
        let (h0, h1, _) = SocketDispatcher::create(0).expect("failed to create stream socket");
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        let (_m_mem, m_ptr) = make_user_in(b"hello").unwrap();
        let w = d0.write(m_ptr, 5).expect("write initial data");
        expect_eq!(w, 5);

        // Disable writes on d0.
        expect_ok!(d0.set_disposition(Disposition::WriteDisabled, Disposition::None));

        // d0 should have ZX_SOCKET_WRITE_DISABLED asserted and ZX_SOCKET_WRITABLE cleared.
        // d1 should have ZX_SOCKET_PEER_WRITE_DISABLED asserted.
        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_true!((sigs0 & ZX_SOCKET_WRITE_DISABLED) != 0);
            expect_eq!(sigs0 & ZX_SOCKET_WRITABLE, 0);
        }
        {
            ksync::lock!(let guard1 = d1.state().peered.lock());
            let sigs1 = d1.signals_state_locked(guard1.token());
            expect_true!((sigs1 & ZX_SOCKET_PEER_WRITE_DISABLED) != 0);
        }

        // Writing on write-disabled socket fails with BAD_STATE.
        let res_w = d0.write(m_ptr, 5);
        expect_true!(res_w == Err(Status::BAD_STATE));

        // Re-enabling writes while unread buffered data is present must fail with BAD_STATE.
        let res_re_enable = d0.set_disposition(Disposition::WriteEnabled, Disposition::None);
        expect_true!(res_re_enable == Err(Status::BAD_STATE));

        // Read out the 5 bytes from d1.
        let (_out_mem, out_ptr) = make_user_out(5).unwrap();
        let nread = d1.read(ReadType::Consume, out_ptr, 5).expect("read buffered data");
        expect_eq!(nread, 5);

        // Now that the buffer is empty, reading again with peer write disabled returns BAD_STATE.
        let res_empty_read = d1.read(ReadType::Consume, out_ptr, 1);
        expect_true!(res_empty_read == Err(Status::BAD_STATE));

        // Now that buffer is empty, re-enabling writes on d0 succeeds.
        expect_ok!(d0.set_disposition(Disposition::WriteEnabled, Disposition::None));

        // d0 should now be writable again.
        {
            ksync::lock!(let guard0 = d0.state().peered.lock());
            let sigs0 = d0.signals_state_locked(guard0.token());
            expect_true!((sigs0 & ZX_SOCKET_WRITABLE) != 0);
            expect_eq!(sigs0 & ZX_SOCKET_WRITE_DISABLED, 0);
        }

        // Writing to d0 succeeds again.
        let w_again = d0.write(m_ptr, 5).expect("write again");
        expect_eq!(w_again, 5);
    }

    /// Tests datagram zero-length, oversize, and capacity limits.
    #[test]
    fn test_datagram_atomic_and_capacity_limits() {
        let (h0, h1, _) =
            SocketDispatcher::create(ZX_SOCKET_DATAGRAM).expect("failed to create datagram socket");
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        let info = d0.get_info();
        let max_capacity = info.rx_buf_max;

        // Writing zero-length datagram via SocketDispatcher::write returns Ok(0).
        let (_zero_mem, zero_ptr) = make_user_in(&[]).unwrap();
        let res_zero = d0.write(zero_ptr, 0);
        expect_true!(matches!(res_zero, Ok(0)));

        // Writing datagram larger than max capacity returns OUT_OF_RANGE.
        let huge_size = max_capacity + 1;
        let (_huge_mem, huge_ptr) = make_user_in_pattern(1, |_| 0).unwrap();
        let res_huge = d0.write(huge_ptr, huge_size);
        expect_true!(res_huge == Err(Status::OUT_OF_RANGE));

        // Test multi-buffer datagram (larger than a single MBuf page).
        const MULTI_PAGE_PAYLOAD: usize = 8192;
        let (_multi_mem, multi_ptr) =
            make_user_in_pattern(MULTI_PAGE_PAYLOAD, |i| (i % 251) as u8).unwrap();
        let w_multi = d0.write(multi_ptr, MULTI_PAGE_PAYLOAD).expect("write multi-page datagram");
        expect_eq!(w_multi, MULTI_PAGE_PAYLOAD);

        let info1 = d1.get_info();
        expect_eq!(info1.rx_buf_available, MULTI_PAGE_PAYLOAD);
        expect_eq!(info1.rx_buf_size, MULTI_PAGE_PAYLOAD);

        let (out_mem, out_ptr) = make_user_out(MULTI_PAGE_PAYLOAD).unwrap();
        let nread =
            d1.read(ReadType::Consume, out_ptr, MULTI_PAGE_PAYLOAD).expect("read multi-page");
        expect_eq!(nread, MULTI_PAGE_PAYLOAD);
        let b = read_user_mem(&out_mem, MULTI_PAGE_PAYLOAD).unwrap();
        for (i, &val) in b.iter().enumerate() {
            expect_eq!(val, (i % 251) as u8);
        }
    }

    /// Tests repeated multi-buffer stream writes and partial reads that drop MBuf pages during read.
    #[test]
    fn test_stream_read_multi_buffer_drop_and_reuse() {
        let (h0, h1, _) = SocketDispatcher::create(0).expect("create stream socket");
        let d0 = h0.dispatcher();
        let d1 = h1.dispatcher();

        const PAYLOAD: usize = 16 * 1024;
        let (_in_mem, in_ptr) = make_user_in_pattern(PAYLOAD, |i| (i & 0xff) as u8).unwrap();
        let (out_mem, out_ptr) = make_user_out(PAYLOAD).unwrap();

        for _ in 0..5 {
            let w = d0.write(in_ptr, PAYLOAD).expect("write stream");
            expect_eq!(w, PAYLOAD);

            let mut total_read = 0usize;
            while total_read < PAYLOAD {
                let chunk = core::cmp::min(1024, PAYLOAD - total_read);
                let r = d1
                    .read(ReadType::Consume, out_ptr.byte_offset(total_read as isize), chunk)
                    .expect("read stream chunk");
                expect_eq!(r, chunk);
                total_read += r;
            }

            let b = read_user_mem(&out_mem, PAYLOAD).unwrap();
            for (i, &val) in b.iter().enumerate() {
                expect_eq!(val, (i & 0xff) as u8);
            }
        }
    }
}