hrx/fabric/

queue.rs

mod sdma;
use super::*;
use std::{
    collections::VecDeque,
    sync::{
        Mutex, Weak,
        atomic::{AtomicU32, AtomicU64, Ordering},
    },
    time::{Duration, Instant},
};

/// A serialized host producer for one native GPU PM4 queue.
#[derive(Clone)]
pub struct Queue(Arc<QueueInner>);

// Weak slots avoid a device -> queue -> device ownership cycle. Streams,
// prepared commands and completions keep their assigned queue alive.
pub(super) struct StreamQueues {
    slots: Vec<Weak<QueueInner>>,
    limit: usize,
    next: usize,
}
impl Default for StreamQueues {
    fn default() -> Self {
        Self {
            slots: Vec::new(),
            // See benchmarks/queue-pool for the gfx1151 throughput/latency tradeoff.
            limit: 8,
            next: 0,
        }
    }
}
impl StreamQueues {
    fn set_limit(&mut self, count: usize) -> Result<()> {
        if count == 0 {
            return Err(Error::Message("stream queue count must be positive".into()));
        }
        if count != self.limit && !self.slots.is_empty() {
            return Err(Error::Busy(
                "configure stream queues before creating the first stream".into(),
            ));
        }
        self.limit = count;
        Ok(())
    }
}
struct QueueInner {
    device: Device,
    family: u32,
    raw: *mut amdf_user_queue_t,
    mapping: *mut amdf_user_queue_mapping_t,
    info: amdf_user_queue_mapping_info_t,
    state: Mutex<QueueState>,
    transfers: Mutex<Option<[Kernel; 4]>>,
}
struct QueueState {
    write: u64,
    pending: VecDeque<Pending>,
}
struct Pending {
    work: Arc<Work>,
    value: u32,
}
struct Work {
    fence: Buffer,
    _kernels: Vec<Kernel>,
    buffers: Vec<Buffer>,
    leases: Mutex<Vec<memory::DeviceUse>>,
    retired: AtomicU32,
    active: AtomicU32,
    dependencies: Vec<(Arc<Work>, u32)>,
}
// Command publication has one mutex-protected producer. Status observation is
// documented as concurrent; mapped native pointers remain valid until drop.
unsafe impl Send for QueueInner {}
unsafe impl Sync for QueueInner {}

/// Native execution completion retaining all indirect resources.
#[derive(Clone)]
pub struct Completion {
    queue: Queue,
    work: Arc<Work>,
    value: u32,
}
impl Completion {
    /// Read cached retirement without native queries, locks, or progress work.
    /// Call `refresh` or `wait` to advance retirement.
    pub fn is_complete(&self) -> Result<bool> {
        Ok(self.work.retired.load(Ordering::Acquire) >= self.value)
    }
    /// Observe the execution fence and retire completed resource leases.
    /// This performs bounded progress work; ring consumption is not completion.
    pub fn refresh(&self) -> Result<bool> {
        if !self.queue.0.poll(&self.work, self.value)? {
            return Ok(false);
        }
        // A completed prefix must release every earlier submission's leases,
        // even when callers observed only the last completion point.
        let mut state = self
            .queue
            .0
            .state
            .lock()
            .map_err(|_| Error::DeviceLost("queue producer poisoned".into()))?;
        while let Some(work) = state.pending.front() {
            if !self.queue.0.poll(&work.work, work.value)? {
                break;
            }
            state.pending.pop_front();
        }
        Ok(true)
    }
    /// Wait without cancelling or releasing pending resources on timeout.
    pub fn wait_timeout(&self, timeout: Duration) -> Result<bool> {
        let start = Instant::now();
        loop {
            if self.refresh()? {
                return Ok(true);
            }
            if start.elapsed() >= timeout {
                return Ok(false);
            }
            std::thread::yield_now();
        }
    }
    /// Wait until execution retires or the provider reports failure.
    pub fn wait(&self) -> Result<()> {
        while !self.refresh()? {
            std::thread::yield_now();
        }
        Ok(())
    }
}
impl Work {
    fn retire(&self, value: u32) -> Result<bool> {
        if self.retired.load(Ordering::Acquire) >= value {
            return Ok(true);
        }
        let mut leases = self
            .leases
            .lock()
            .map_err(|_| Error::DeviceLost("submission ownership poisoned".into()))?;
        if self.retired.load(Ordering::Acquire) >= value {
            return Ok(true);
        }
        let completed = self.fence.fence_value()?;
        if completed >= value {
            for (dependency, point) in &self.dependencies {
                if !dependency.retire(*point)? {
                    return Ok(false);
                }
            }
            if completed >= self.active.load(Ordering::Acquire) {
                leases.clear();
            }
            self.retired.store(completed, Ordering::Release);
            return Ok(true);
        }
        Ok(false)
    }
}
impl QueueInner {
    fn poll(&self, work: &Work, value: u32) -> Result<bool> {
        if work.retire(value)? {
            return Ok(true);
        }
        let api = self.device.0.endpoint.0.instance.api.core;
        let mut status = amdf_user_queue_status_t {
            type_: AMDF_STRUCTURE_TYPE_USER_QUEUE_STATUS,
            structure_size: size_of::<amdf_user_queue_status_t>() as u32,
            ..Default::default()
        };
        check("user_queue_query_status", unsafe {
            entry!(api, user_queue_query_status)(self.raw, &mut status)
        })?;
        if status.state != AMDF_QUEUE_STATE_ACTIVE {
            return Err(Error::DeviceLost(format!(
                "PM4 queue failed with native status {}",
                status.terminal_status
            )));
        }
        Ok(false)
    }
}
impl Drop for QueueInner {
    fn drop(&mut self) {
        let pending = std::mem::take(
            &mut self
                .state
                .get_mut()
                .unwrap_or_else(|e| e.into_inner())
                .pending,
        );
        let deadline = Instant::now() + Duration::from_secs(10);
        for work in &pending {
            loop {
                match self.poll(&work.work, work.value) {
                    Ok(true) => break,
                    Ok(false) if Instant::now() < deadline => std::thread::yield_now(),
                    Ok(false) | Err(_) => {
                        // Failure is not retirement. Leak the entire native
                        // ownership chain instead of unmapping reachable memory.
                        std::mem::forget(pending);
                        std::mem::forget(self.device.clone());
                        return;
                    }
                }
            }
        }
        let api = self.device.0.endpoint.0.instance.api.core;
        unsafe {
            if !self.mapping.is_null() && api.user_queue_mapping_destroy.unwrap()(self.mapping) != 0
            {
                std::mem::forget(self.device.clone());
                return;
            }
            if !self.raw.is_null() && api.user_queue_destroy.unwrap()(self.raw) != 0 {
                std::mem::forget(self.device.clone());
            }
        }
    }
}

fn header(opcode: u32, count: usize) -> u32 {
    (3 << 30) | (((count - 2) as u32) << 16) | (opcode << 8)
}
fn barrier(words: &mut Vec<u32>) {
    // Native PM4 v1/GCR system release and acquire. GLI_INV=1 invalidates
    // every instruction-cache line; 3 means FIRST_LAST, not ALL. A full address
    // range with FIRST_LAST can retain instructions from retired code whose
    // allocation address is later reused. See Mesa's GCR_GLI_INV packet enum
    // (src/amd/registers/pkt3.json) and PAL's BuildAcquireMemInternal.
    const GLI_INV_ALL: u32 = 1;
    words.extend([
        header(0x46, 2),
        7 | (4 << 8),
        header(0x58, 8),
        0,
        u32::MAX,
        0xff,
        0,
        0,
        0x0a,
        GLI_INV_ALL | (1 << 4) | (1 << 5) | (1 << 7) | (1 << 8) | (1 << 9) | (1 << 14) | (1 << 15),
    ]);
}
fn pad(words: &mut Vec<u32>) {
    let mut count = 8 - words.len() % 8;
    if count == 1 {
        count += 8;
    }
    words.push(header(0x10, count));
    words.resize(words.len() + count - 1, 0);
}

// The native indirect-buffer size field is 20 bits. Padding needs up to nine
// words; dependency barriers take ten and each optional marker at most 32.
const INDIRECT_WORD_LIMIT: usize = 1 << 20;
fn batch_ranges(
    commands: impl IntoIterator<Item = (usize, bool)>,
    profiled: bool,
) -> Result<Vec<std::ops::Range<usize>>> {
    let mut ranges = Vec::new();
    let mut start = 0;
    let mut end = 0;
    let mut words = 0;
    for (payload, dependency) in commands {
        let size = payload
            .checked_add(if dependency { 10 } else { 0 })
            .and_then(|n| n.checked_add(if profiled { 2 * (10 + 32) } else { 0 }))
            .filter(|n| *n < INDIRECT_WORD_LIMIT - 9)
            .ok_or_else(|| {
                Error::Unsupported("native command exceeds indirect command capacity".into())
            })?;
        if words + size >= INDIRECT_WORD_LIMIT - 9 {
            ranges.push(start..end);
            start = end;
            words = 0;
        }
        words += size;
        end += 1;
    }
    if start != end {
        ranges.push(start..end);
    }
    Ok(ranges)
}

impl Device {
    pub(crate) fn set_stream_queue_count(&self, count: usize) -> Result<()> {
        self.0
            .stream_queues
            .lock()
            .map_err(|_| Error::DeviceLost("stream queue pool poisoned".into()))?
            .set_limit(count)
    }

    pub(crate) fn stream_queue_count(&self) -> Result<usize> {
        Ok(self
            .0
            .stream_queues
            .lock()
            .map_err(|_| Error::DeviceLost("stream queue pool poisoned".into()))?
            .limit)
    }

    pub(crate) fn stream_queue(&self) -> Result<Queue> {
        let mut pool = self
            .0
            .stream_queues
            .lock()
            .map_err(|_| Error::DeviceLost("stream queue pool poisoned".into()))?;
        let slot = pool.next;
        if slot == pool.slots.len() {
            pool.slots
                .try_reserve(1)
                .map_err(|_| Error::Message("stream queue pool allocation failed".into()))?;
            let queue = self.queue()?;
            pool.slots.push(Arc::downgrade(&queue.0));
            pool.next = (slot + 1) % pool.limit;
            return Ok(queue);
        }
        let queue = match pool.slots[slot].upgrade() {
            Some(queue) => Queue(queue),
            None => {
                let queue = self.queue()?;
                pool.slots[slot] = Arc::downgrade(&queue.0);
                queue
            }
        };
        pool.next = (slot + 1) % pool.limit;
        Ok(queue)
    }

    /// Create a native PM4 queue with one serialized host producer.
    pub fn queue(&self) -> Result<Queue> {
        self.create_queue(AMDF_QUEUE_COMMAND_TYPE_GPU_PM4)
    }
    /// Create a native SDMA transfer queue without loading compute kernels.
    pub fn sdma_queue(&self) -> Result<Queue> {
        self.create_queue(AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA)
    }
    fn create_queue(&self, command: u32) -> Result<Queue> {
        if self.endpoint().engine() != Engine::Gpu {
            return Err(Error::Unsupported(
                "PM4 queues require an AMDGPU device".into(),
            ));
        }
        let api = &self.0.endpoint.0.instance.api;
        let _ = entry!(api.core, user_queue_destroy);
        let _ = entry!(api.core, user_queue_mapping_destroy);
        let mut endpoint = amdf_endpoint_info_t {
            type_: AMDF_STRUCTURE_TYPE_ENDPOINT_INFO,
            structure_size: size_of::<amdf_endpoint_info_t>() as u32,
            ..Default::default()
        };
        check("endpoint_query_info", unsafe {
            entry!(api.core, endpoint_query_info)(self.0.endpoint.0.raw, &mut endpoint)
        })?;
        for ordinal in 0..endpoint.queue_family_count {
            let mut family = amdf_queue_family_info_t {
                type_: AMDF_STRUCTURE_TYPE_QUEUE_FAMILY_INFO,
                structure_size: size_of::<amdf_queue_family_info_t>() as u32,
                ..Default::default()
            };
            check("endpoint_query_queue_family_info", unsafe {
                entry!(api.core, endpoint_query_queue_family_info)(
                    self.0.endpoint.0.raw,
                    ordinal,
                    &mut family,
                )
            })?;
            if family.command_type != command
                || family.format_version != AMDF_GPU_PM4_QUEUE_FORMAT_VERSION_1
                || family.publication_modes & AMDF_QUEUE_PUBLICATION_MODE_USER == 0
                || (command == AMDF_QUEUE_COMMAND_TYPE_GPU_PM4
                    && family.format_features & AMDF_GPU_PM4_FORMAT_FEATURE_ACQUIRE_MEM_GCR as u64
                        == 0)
            {
                continue;
            }
            let options = amdf_gpu_user_queue_create_info_t {
                type_: HRX_AMDF_STRUCTURE_TYPE_GPU_USER_QUEUE_CREATE_INFO,
                structure_size: size_of::<amdf_gpu_user_queue_create_info_t>() as u32,
                queue_family_ordinal: ordinal,
                priority: AMDF_QUEUE_PRIORITY_NORMAL,
                producer_mode: AMDF_QUEUE_PRODUCER_MODE_SINGLE,
                required_capabilities: AMDF_USER_QUEUE_CAPABILITY_HOST_PRODUCER as u64,
                ring_byte_length: family.minimum_ring_byte_length,
                ..Default::default()
            };
            let mut raw = ptr::null_mut();
            check("gpu.user_queue_create", unsafe {
                entry!(api.gpu, user_queue_create)(self.0.raw, &options, &mut raw)
            })?;
            if raw.is_null() {
                return Err(missing("created queue"));
            }
            let mut queue = QueueInner {
                device: self.clone(),
                family: ordinal,
                raw,
                mapping: ptr::null_mut(),
                info: amdf_user_queue_mapping_info_t::default(),
                transfers: Mutex::new(None),
                state: Mutex::new(QueueState {
                    write: 0,
                    pending: VecDeque::with_capacity(4096),
                }),
            };
            check("user_queue_map", unsafe {
                entry!(api.core, user_queue_map)(raw, ptr::null_mut(), &mut queue.mapping)
            })?;
            queue.info.type_ = AMDF_STRUCTURE_TYPE_USER_QUEUE_MAPPING_INFO;
            queue.info.structure_size = size_of::<amdf_user_queue_mapping_info_t>() as u32;
            check("user_queue_mapping_query_info", unsafe {
                entry!(api.core, user_queue_mapping_query_info)(queue.mapping, &mut queue.info)
            })?;
            let info = &queue.info;
            if info.command_type != command
                || info.format_version != 1
                || !info.ring_byte_length.is_power_of_two()
                || info.ring_byte_length < 4096
                || info.index_bits != 64
                || info.doorbell_bits != 64
                || [
                    info.ring_address,
                    info.read_index_address,
                    info.write_index_address,
                    info.doorbell_address,
                ]
                .iter()
                .any(|address| *address == 0 || address % 8 != 0)
            {
                return Err(Error::Unsupported(
                    "unexpected native GPU queue mapping contract".into(),
                ));
            }
            return Ok(Queue(Arc::new(queue)));
        }
        Err(Error::Unsupported(
            "no qualified native queue family for the requested engine".into(),
        ))
    }
}

/// Private graph construction storage. Nodes are never submitted individually.
/// Argument pages become immutable before the graph can be launched.
#[derive(Default)]
pub(crate) struct GraphArena {
    page: Option<Buffer>,
    used: usize,
    fence: Option<Buffer>,
}
impl GraphArena {
    fn pack(&mut self, queue: &Queue, bytes: &[u8]) -> Result<(Buffer, usize, Buffer, usize)> {
        let fabric = queue.device().fabric();
        let size = bytes.len().max(64).next_multiple_of(64);
        let mut allocated = 0;
        if self
            .page
            .as_ref()
            .is_none_or(|p| size > p.len() - self.used)
        {
            self.page =
                Some(fabric.allocate(size.max(64 * 1024), std::slice::from_ref(queue.device()))?);
            self.used = 0;
            allocated += self.page.as_ref().unwrap().len();
        }
        let page = self.page.as_ref().unwrap();
        let offset = self.used;
        page.write(offset, bytes)?;
        self.used += size;
        if self.fence.is_none() {
            self.fence = Some(fabric.allocate_shared(64, std::slice::from_ref(queue.device()))?);
            allocated += 64;
        }
        Ok((
            page.clone(),
            offset,
            self.fence.as_ref().unwrap().clone(),
            allocated,
        ))
    }
}

impl Queue {
    pub(super) fn identity(&self) -> usize {
        Arc::as_ptr(&self.0) as usize
    }
    /// Prepare immutable commands and backing for repeated native execution.
    ///
    /// # Safety
    /// Arguments and launch dimensions must satisfy the kernel's memory
    /// contract. The caller orders conflicting accesses across other queues.
    pub unsafe fn prepare(
        &self,
        kernel: &Kernel,
        grid: [u32; 3],
        block: [u16; 3],
        arguments: &[Argument<'_>],
    ) -> Result<PreparedGpu> {
        unsafe { self.prepare_inner(kernel, grid, block, arguments, None) }
    }
    pub(crate) unsafe fn prepare_graph(
        &self,
        kernel: &Kernel,
        grid: [u32; 3],
        block: [u16; 3],
        arguments: &[Argument<'_>],
        arena: &mut GraphArena,
    ) -> Result<PreparedGpu> {
        unsafe { self.prepare_inner(kernel, grid, block, arguments, Some(arena)) }
    }
    unsafe fn prepare_inner(
        &self,
        kernel: &Kernel,
        grid: [u32; 3],
        block: [u16; 3],
        arguments: &[Argument<'_>],
        arena: Option<&mut GraphArena>,
    ) -> Result<PreparedGpu> {
        if self.0.info.command_type != AMDF_QUEUE_COMMAND_TYPE_GPU_PM4 {
            return Err(Error::Unsupported(
                "kernel dispatch requires a compute queue".into(),
            ));
        }
        if !Arc::ptr_eq(&kernel.device().0, &self.0.device.0) {
            return Err(Error::Message(
                "kernel belongs to another native device".into(),
            ));
        }
        let fabric = Fabric(self.0.device.0.endpoint.0.instance.clone());
        let (packed, mut buffers) = kernel.arguments(arguments)?;
        let (argument_buffer, argument_offset, fence, argument_storage) = if let Some(arena) = arena
        {
            arena.pack(self, &packed)?
        } else {
            let buffer =
                fabric.allocate(packed.len().max(64), std::slice::from_ref(&self.0.device))?;
            buffer.write(0, &packed)?;
            {
                let bytes = buffer.len() + 64;
                (
                    buffer,
                    0,
                    fabric.allocate_shared(64, std::slice::from_ref(&self.0.device))?,
                    bytes,
                )
            }
        };
        let address = fence.device_address(&self.0.device)?;
        let scratch = if kernel.0.info.private_bytes != 0 {
            let mut info = amdf_gpu_endpoint_info_t {
                type_: HRX_AMDF_STRUCTURE_TYPE_GPU_ENDPOINT_INFO,
                structure_size: size_of::<amdf_gpu_endpoint_info_t>() as u32,
                ..Default::default()
            };
            let api = &self.0.device.0.endpoint.0.instance.api;
            check("GPU scratch topology", unsafe {
                entry!(api.gpu, endpoint_query_info)(self.0.device.0.endpoint.0.raw, &mut info)
            })?;
            if info.topology.xcc_count != 1 {
                return Err(Error::Unsupported(
                    "scratch requires the qualified single-XCC topology".into(),
                ));
            }
            let waves = info
                .compute
                .compute_unit_count
                .checked_mul(info.compute.maximum_scratch_wave_count_per_compute_unit)
                .ok_or_else(|| Error::Message("scratch topology overflow".into()))?;
            let per_wave = (u64::from(kernel.0.info.private_bytes)
                * u64::from(kernel.0.info.wave_size))
            .div_ceil(256)
                * 256;
            let size = per_wave
                .checked_mul(u64::from(waves))
                .filter(|size| *size <= 512 * 1024 * 1024)
                .ok_or_else(|| {
                    Error::Unsupported("kernel scratch exceeds 512 MiB admission limit".into())
                })?;
            let buffer = fabric.allocate_access(
                size as usize,
                std::slice::from_ref(&self.0.device),
                AMDF_MEMORY_ACCESS_READ | AMDF_MEMORY_ACCESS_WRITE,
                256,
            )?;
            Some((buffer, waves, info.topology.shader_engine_count_per_xcc))
        } else {
            None
        };
        let mut words = Vec::with_capacity(160);
        barrier(&mut words);
        let dispatch_start = words.len();
        words.extend(
            kernel.dispatch_words(
                grid,
                block,
                &packed,
                argument_buffer.device_address(&self.0.device)? + argument_offset as u64,
                scratch
                    .as_ref()
                    .map(|(buffer, waves, engines)| (buffer, *waves, *engines)),
            )?,
        );
        let dispatch_end = words.len();
        barrier(&mut words);
        let fence_word = words.len() + 4;
        words.extend([
            header(0x37, 5),
            (2 << 8) | (1 << 20),
            address as u32,
            (address >> 32) as u32,
            1,
        ]);
        pad(&mut words);
        // Arena pages are charged once, at their first graph node.
        let storage_bytes =
            argument_storage + scratch.as_ref().map_or(0, |(buffer, _, _)| buffer.len());
        buffers.push(argument_buffer.clone());
        if let Some((scratch, _, _)) = scratch {
            buffers.push(scratch);
        }
        buffers.sort_by_key(|buffer| Arc::as_ptr(&buffer.0) as usize);
        buffers.dedup_by(|a, b| Arc::ptr_eq(&a.0, &b.0));
        let leases = Mutex::new(Vec::with_capacity(buffers.len()));
        Ok(PreparedGpu {
            prefix: Default::default(),
            queue: self.clone(),
            inner: Arc::new(PreparedInner {
                storage_bytes,
                work: Arc::new(Work {
                    _kernels: vec![kernel.clone()],
                    buffers,
                    fence,
                    leases,
                    retired: AtomicU32::new(0),
                    active: AtomicU32::new(0),
                    dependencies: Vec::new(),
                }),
                words,
                dispatch_range: dispatch_start..dispatch_end,
                indirect_words: None,
                fence_word,
                previous: Mutex::new(0),
            }),
        })
    }
    /// Combine prepared dispatches into one immutable indirect command buffer.
    /// A true flag establishes a system-memory dependency on preceding nodes.
    ///
    /// # Safety
    /// The supplied barriers must order every conflicting access. Commands must
    /// obey their kernel contracts and must not use this batch's storage elsewhere.
    pub unsafe fn prepare_batch(&self, commands: &[(PreparedGpu, bool)]) -> Result<PreparedGpu> {
        self.prepare_batch_inner(commands, None, 0)
    }
    /// Instrument every prepared command, including nested batches as one span.
    /// Completion barriers serialize the sampled commands. This is diagnostic
    /// execution; use ordinary batches for end-to-end performance decisions.
    ///
    /// # Safety
    /// The same kernel and cross-queue contracts as `prepare_batch` apply.
    pub unsafe fn prepare_profiled_batch(
        &self,
        commands: &[(PreparedGpu, bool)],
        labels: &[String],
    ) -> Result<ProfiledGpu> {
        if commands.len() != labels.len() {
            return Err(Error::Message("profile labels must match commands".into()));
        }
        let capture = profile::ProfileCapture::new(self, labels)?;
        let commands = self.prepare_batch_inner(commands, Some(&capture), 0)?;
        Ok(ProfiledGpu { commands, capture })
    }
    /// Prepare a graph as bounded submissions on this ordered queue. Build all
    /// batches before launching any, retaining one shared timestamp capture.
    /// Safety: the same kernel/access contracts as `prepare_batch` apply.
    pub(crate) unsafe fn prepare_graph_batches(
        &self,
        commands: &[(PreparedGpu, bool)],
        labels: Option<&[String]>,
    ) -> Result<(Vec<PreparedGpu>, Option<profile::ProfileCapture>)> {
        let capture = if let Some(labels) = labels {
            if commands.len() != labels.len() || commands.is_empty() {
                return Err(Error::Message(
                    "profile labels must match nonempty commands".into(),
                ));
            }
            Some(profile::ProfileCapture::new(self, labels)?)
        } else {
            None
        };
        let ranges = batch_ranges(
            commands
                .iter()
                .map(|(command, dependency)| (command.dispatch_words().len(), *dependency)),
            capture.is_some(),
        )?;
        let batches = ranges
            .into_iter()
            .map(|range| {
                self.prepare_batch_inner(&commands[range.clone()], capture.as_ref(), range.start)
            })
            .collect::<Result<_>>()?;
        Ok((batches, capture))
    }
    fn prepare_batch_inner(
        &self,
        commands: &[(PreparedGpu, bool)],
        profile: Option<&profile::ProfileCapture>,
        profile_start: usize,
    ) -> Result<PreparedGpu> {
        if self.0.info.command_type != AMDF_QUEUE_COMMAND_TYPE_GPU_PM4 {
            return Err(Error::Unsupported(
                "indirect compute batches require a PM4 queue".into(),
            ));
        }
        if commands.is_empty() {
            return Err(Error::Message("empty native command batch".into()));
        }
        let mut indirect = Vec::new();
        let mut buffers = Vec::new();
        let mut kernels = Vec::new();
        let mut dependencies = Vec::new();
        let timestamp_address = profile
            .map(|p| p.buffer.device_address(self.device()))
            .transpose()?;
        if let Some(profile) = profile {
            buffers.push(profile.buffer.clone());
        }
        for (index, (command, dependency)) in commands.iter().enumerate() {
            if !Arc::ptr_eq(&command.queue.0, &self.0) {
                return Err(Error::Message(
                    "batch command belongs to another queue".into(),
                ));
            }
            if *dependency {
                barrier(&mut indirect);
            }
            if let (Some(profile), Some(address)) = (profile, timestamp_address) {
                barrier(&mut indirect);
                profile
                    .api
                    .emit(address + (profile_start + index) as u64 * 16, &mut indirect)?;
            }
            indirect.extend_from_slice(command.dispatch_words());
            if let (Some(profile), Some(address)) = (profile, timestamp_address) {
                barrier(&mut indirect);
                profile.api.emit(
                    address + (profile_start + index) as u64 * 16 + 8,
                    &mut indirect,
                )?;
            }
            buffers.extend(command.inner.work.buffers.iter().cloned());
            kernels.extend(command.inner.work._kernels.iter().cloned());
            dependencies.extend(command.inner.work.dependencies.iter().cloned());
        }
        pad(&mut indirect);
        if indirect.len() >= INDIRECT_WORD_LIMIT {
            return Err(Error::Unsupported(format!(
                "native batch exceeds indirect command capacity: {} words for {} commands",
                indirect.len(),
                commands.len()
            )));
        }
        let fabric = self.device().fabric();
        let storage = fabric.allocate_access(
            indirect.len() * 4,
            std::slice::from_ref(self.device()),
            AMDF_MEMORY_ACCESS_READ | AMDF_MEMORY_ACCESS_WRITE | AMDF_MEMORY_ACCESS_EXECUTE,
            256,
        )?;
        // Native PM4 words are little endian on this supported x86_64 host.
        storage.write(0, unsafe {
            std::slice::from_raw_parts(indirect.as_ptr().cast(), indirect.len() * 4)
        })?;
        let ib_address = storage.device_address(self.device())?;
        let storage_bytes = storage.len()
            + 64
            + profile.map_or(0, |capture| capture.buffer.len())
            + commands
                .iter()
                .map(|(command, _)| command.storage_bytes())
                .sum::<usize>();
        buffers.push(storage);
        buffers.sort_by_key(|buffer| Arc::as_ptr(&buffer.0) as usize);
        buffers.dedup_by(|a, b| Arc::ptr_eq(&a.0, &b.0));
        let fence = fabric.allocate_shared(64, std::slice::from_ref(self.device()))?;
        let address = fence.device_address(self.device())?;
        let mut words = Vec::new();
        barrier(&mut words);
        let dispatch_start = words.len();
        words.extend([
            header(0x3f, 4),
            ib_address as u32,
            (ib_address >> 32) as u32,
            indirect.len() as u32 | (1 << 23),
        ]);
        let dispatch_end = words.len();
        barrier(&mut words);
        let fence_word = words.len() + 4;
        words.extend([
            header(0x37, 5),
            (2 << 8) | (1 << 20),
            address as u32,
            (address >> 32) as u32,
            1,
        ]);
        pad(&mut words);
        let leases = Mutex::new(Vec::with_capacity(buffers.len()));
        Ok(PreparedGpu {
            prefix: Default::default(),
            queue: self.clone(),
            inner: Arc::new(PreparedInner {
                storage_bytes,
                work: Arc::new(Work {
                    fence,
                    _kernels: kernels,
                    buffers,
                    leases,
                    retired: AtomicU32::new(0),
                    active: AtomicU32::new(0),
                    dependencies,
                }),
                words,
                dispatch_range: dispatch_start..dispatch_end,
                indirect_words: Some(indirect),
                fence_word,
                previous: Mutex::new(0),
            }),
        })
    }
    /// Prepare a device-side wait for an immutable completion point.
    /// The source fence and all indirect resources remain owned through retirement.
    pub fn prepare_wait(&self, completion: &Completion) -> Result<PreparedGpu> {
        if self.device().id() != completion.queue.device().id() {
            return Err(Error::Message("event belongs to another device".into()));
        }
        // The event orders the entire producer prefix, not just the buffers in
        // its final command. Retire those leases when a consumer observes the
        // wait, even if nobody polls the producer again. Snapshot work rather
        // than retaining/locking its queue during retirement: queues may wait
        // on their own earlier events, and cross-queue waits may form a chain.
        let dependencies = {
            let state = completion
                .queue
                .0
                .state
                .lock()
                .map_err(|_| Error::DeviceLost("event producer poisoned".into()))?;
            match state.pending.iter().position(|pending| {
                Arc::ptr_eq(&pending.work, &completion.work) && pending.value == completion.value
            }) {
                Some(end) => state
                    .pending
                    .iter()
                    .take(end + 1)
                    .map(|pending| (pending.work.clone(), pending.value))
                    .collect(),
                None => vec![(completion.work.clone(), completion.value)],
            }
        };
        if self.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA {
            return self.prepare_sdma_wait(completion, dependencies);
        }
        let address = completion.work.fence.device_address(self.device())?;
        let fence = self
            .device()
            .fabric()
            .allocate_shared(64, std::slice::from_ref(self.device()))?;
        let output = fence.device_address(self.device())?;
        let mut words = Vec::with_capacity(40);
        // WAIT_REG_MEM32, memory space, unsigned >= comparison, MEC polling.
        words.extend([
            header(0x3c, 7),
            5 | (1 << 4),
            address as u32,
            (address >> 32) as u32,
            completion.value,
            u32::MAX,
            4,
        ]);
        barrier(&mut words);
        let dispatch_end = words.len();
        let fence_word = words.len() + 4;
        words.extend([
            header(0x37, 5),
            (2 << 8) | (1 << 20),
            output as u32,
            (output >> 32) as u32,
            1,
        ]);
        pad(&mut words);
        Ok(PreparedGpu {
            prefix: Default::default(),
            queue: self.clone(),
            inner: Arc::new(PreparedInner {
                storage_bytes: fence.len(),
                work: Arc::new(Work {
                    fence,
                    _kernels: Vec::new(),
                    buffers: Vec::new(),
                    leases: Mutex::new(Vec::new()),
                    retired: AtomicU32::new(0),
                    active: AtomicU32::new(0),
                    dependencies,
                }),
                words,
                dispatch_range: 0..dispatch_end,
                indirect_words: None,
                fence_word,
                previous: Mutex::new(0),
            }),
        })
    }
    /// Prepare and publish a single invocation.
    ///
    /// # Safety
    /// The kernel and its arguments must obey the declared memory contract;
    /// callers order conflicting accesses on other queues.
    pub unsafe fn dispatch(
        &self,
        kernel: &Kernel,
        grid: [u32; 3],
        block: [u16; 3],
        arguments: &[Argument<'_>],
    ) -> Result<Completion> {
        unsafe { self.prepare(kernel, grid, block, arguments)?.dispatch() }
    }
}

/// Immutable bound GPU commands with reusable argument, scratch and fence storage.
/// Replays remain ordered on the originating queue, including scratch accesses.
#[derive(Clone)]
pub struct PreparedGpu {
    // SDMA streams must fit the ring, including wrap padding. The final part
    // supplies the public completion; preceding parts retire on the same queue.
    prefix: Option<Arc<[PreparedGpu]>>,
    queue: Queue,
    inner: Arc<PreparedInner>,
}
struct PreparedInner {
    storage_bytes: usize,
    work: Arc<Work>,
    words: Vec<u32>,
    dispatch_range: std::ops::Range<usize>,
    // Flatten nested batches when composing graphs; PM4 IB nesting is bounded.
    indirect_words: Option<Vec<u32>>,
    fence_word: usize,
    previous: Mutex<u32>,
}
impl PreparedGpu {
    fn dispatch_words(&self) -> &[u32] {
        self.inner
            .indirect_words
            .as_deref()
            .unwrap_or_else(|| &self.inner.words[self.inner.dispatch_range.clone()])
    }

    fn reserve_fence(&mut self, reservation: Option<Arc<crate::residency::MemoryReservation>>) {
        let inner = Arc::get_mut(&mut self.inner).expect("new prepared transfer");
        let work = Arc::get_mut(&mut inner.work).expect("new transfer work");
        Arc::get_mut(&mut work.fence.0)
            .expect("new transfer fence")
            .reservation = reservation;
    }

    /// Keep operands unavailable to host mappings between SDMA publications.
    pub(crate) fn retain_parts(&self) -> Result<impl Drop> {
        if self.prefix.is_none() {
            return Ok(Vec::new());
        }
        self.inner
            .work
            .buffers
            .iter()
            .map(Buffer::retain_use)
            .collect::<Result<Vec<_>>>()
    }

    #[cfg(test)]
    pub(crate) fn exhaust_final_timeline(&self) {
        *self.inner.previous.lock().unwrap() = u32::MAX;
    }

    /// Private argument, scratch and fence bytes, excluding operands, code and
    /// provider allocation rounding. Used to bound retained command caches.
    pub(crate) fn storage_bytes(&self) -> usize {
        self.inner.storage_bytes
            + self.prefix.as_ref().map_or(0, |parts| {
                parts.iter().map(Self::storage_bytes).sum::<usize>()
            })
    }

    pub(crate) fn parts(&self) -> impl Iterator<Item = &Self> {
        self.prefix
            .iter()
            .flat_map(|parts| parts.iter())
            .chain(std::iter::once(self))
    }

    /// Submit one physical part; callers retain the logical transfer's guard
    /// and record each accepted completion before advancing to its next part.
    pub(crate) unsafe fn dispatch_part_wait(&self) -> Result<Completion> {
        unsafe { self.dispatch_one(true) }
    }

    /// Submit previously prepared commands without creating native resources.
    /// Large SDMA transfers may wait for ring capacity between publications.
    /// A failure can leave an accepted prefix modified; it is drained before
    /// returning unless native completion itself fails.
    ///
    /// # Safety
    /// Callers order conflicting accesses on other queues and obey the native
    /// executable's binding contracts.
    pub unsafe fn dispatch(&self) -> Result<Completion> {
        if self.prefix.is_some() {
            unsafe { self.dispatch_wait() }
        } else {
            unsafe { self.dispatch_one(false) }
        }
    }

    /// Wait for ring capacity while publishing all parts. Accepted work is
    /// drained if a later part fails; uncertain completion retains native owners.
    pub(crate) unsafe fn dispatch_wait(&self) -> Result<Completion> {
        let _uses = self.retain_parts()?;
        let mut last: Option<Completion> = None;
        for part in self.parts() {
            match unsafe { part.dispatch_one(true) } {
                Ok(done) => last = Some(done),
                Err(error) => {
                    if let Some(done) = last {
                        done.wait()?;
                    }
                    return Err(error);
                }
            }
        }
        Ok(last.expect("prepared command has a final part"))
    }

    unsafe fn dispatch_one(&self, wait: bool) -> Result<Completion> {
        let mut previous = self
            .inner
            .previous
            .lock()
            .map_err(|_| Error::DeviceLost("prepared command poisoned".into()))?;
        let value = previous
            .checked_add(1)
            .ok_or_else(|| Error::DeviceLost("prepared completion timeline exhausted".into()))?;
        let words = &self.inner.words;
        let capacity = self.queue.0.info.ring_byte_length / 4;
        let sdma = self.queue.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA;
        let index_scale = if sdma { 4 } else { 1 };
        let (mut state, cursor, wrap, required) = loop {
            let mut state = self
                .queue
                .0
                .state
                .lock()
                .map_err(|_| Error::DeviceLost("queue producer poisoned".into()))?;
            // Reap completed submissions even if their public completion was dropped.
            while let Some(work) = state.pending.front() {
                if !self.queue.0.poll(&work.work, work.value)? {
                    break;
                }
                state.pending.pop_front();
            }
            let cursor = state.write & (capacity - 1);
            let wrap = if words.len() as u64 > capacity - cursor {
                capacity - cursor
            } else {
                0
            };
            let required = wrap + words.len() as u64;
            let read = unsafe {
                (&*(self.queue.0.info.read_index_address as *const AtomicU64))
                    .load(Ordering::Acquire)
            } / index_scale;
            let outstanding = state.write.wrapping_sub(read) & (capacity - 1);
            let full = if state.pending.len() >= 4096 {
                Some("native queue submission capacity")
            } else if required >= capacity - outstanding {
                Some("native command ring is full")
            } else {
                None
            };
            let Some(reason) = full else {
                break (state, cursor, wrap, required);
            };
            if !wait {
                return Err(Error::Busy(reason.into()));
            }
            let oldest = state.pending.front().map(|pending| Completion {
                queue: self.queue.clone(),
                work: pending.work.clone(),
                value: pending.value,
            });
            drop(state);
            wait_for_capacity(oldest)?;
        };
        let published = state
            .write
            .checked_add(required)
            .ok_or_else(|| Error::DeviceLost("native queue index exhausted".into()))?;
        let native_published = published
            .checked_mul(index_scale)
            .ok_or_else(|| Error::DeviceLost("native queue byte index exhausted".into()))?;
        let work = self.inner.work.clone();
        let mut leases = work
            .leases
            .lock()
            .map_err(|_| Error::DeviceLost("submission ownership poisoned".into()))?;
        if leases.is_empty() {
            for buffer in &work.buffers {
                match buffer.retain_use() {
                    Ok(lease) => leases.push(lease),
                    Err(error) => {
                        leases.clear();
                        return Err(error);
                    }
                }
            }
        }
        work.active.store(value, Ordering::Release);
        // Fixed capacity and reusable ownership storage are allocated at prepare.
        state.pending.push_back(Pending {
            work: work.clone(),
            value,
        });
        drop(leases);
        unsafe {
            let ring = self.queue.0.info.ring_address as *mut u32;
            if wrap != 0 {
                ptr::write_bytes(ring.add(cursor as usize), 0, wrap as usize);
                if !sdma {
                    ptr::write(ring.add(cursor as usize), header(0x10, wrap as usize));
                }
            }
            let cursor = if wrap == 0 { cursor as usize } else { 0 };
            ptr::copy_nonoverlapping(words.as_ptr(), ring.add(cursor), words.len());
            ptr::write(ring.add(cursor + self.inner.fence_word), value);
            (&*(self.queue.0.info.write_index_address as *const AtomicU64))
                .store(native_published, Ordering::Release);
            std::sync::atomic::fence(Ordering::SeqCst);
            ptr::write_volatile(
                self.queue.0.info.doorbell_address as *mut u64,
                native_published,
            );
        }
        state.write = published;
        *previous = value;
        Ok(Completion {
            queue: self.queue.clone(),
            work,
            value,
        })
    }
}

fn wait_for_capacity(oldest: Option<Completion>) -> Result<()> {
    // A stale native read index or an invalid command must not cause an
    // unbounded spin when there is no submitted work whose fence we can await.
    oldest
        .ok_or_else(|| Error::Busy("native command ring is full without pending work".into()))?
        .wait()
}

impl Queue {
    /// Exact native family used by this queue and its visibility recipes.
    pub fn family_ordinal(&self) -> u32 {
        self.0.family
    }
    /// Native command representation.
    pub fn command(&self) -> QueueCommand {
        if self.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA {
            QueueCommand::Sdma
        } else {
            QueueCommand::Pm4
        }
    }
    /// Device whose address domain this queue executes in.
    pub fn device(&self) -> &Device {
        &self.0.device
    }
    fn transfer_kernels(&self) -> Result<[Kernel; 4]> {
        let mut kernels = self
            .0
            .transfers
            .lock()
            .map_err(|_| Error::DeviceLost("transfer compiler poisoned".into()))?;
        if kernels.is_none() {
            let compiler = crate::loom::Compiler::for_target(None, self.device().target())?;
            let source =
                crate::loom::CxxSource::new("hrx-transfers.cpp", include_str!("transfers.cpp"));
            let module = compiler.import_cxx(source)?;
            // Compile an upper bound for blockIdx; these kernels never read
            // gridDim and guard every byte with the runtime length. Dispatching
            // the checked prefix below therefore preserves their memory contract.
            let request = |symbol: &str| {
                crate::loom::Specialization::new(symbol)
                    .with_config(format!("{symbol}.workgroup_count.x"), "16777216")
                    .with_config(format!("{symbol}.workgroup_count.y"), "1")
                    .with_config(format!("{symbol}.workgroup_count.z"), "1")
            };
            let mut entries = Vec::new();
            for symbol in ["hrx_copy", "hrx_fill", "hrx_copy_words", "hrx_fill_words"] {
                let artifact = module.compile(&request(symbol))?;
                // Crate-owned bounded kernels; prepared calls validate ranges and alignment.
                entries.push(unsafe { self.device().load(&artifact) }?);
            }
            *kernels = Some(entries.try_into().unwrap_or_else(|_| unreachable!()));
        }
        Ok(kernels.as_ref().unwrap().clone())
    }
    /// Prepare a copy of any nonempty, in-bounds, non-overlapping byte range.
    /// Large transfers are split to fit the selected engine. The returned
    /// completion covers every part; operands remain retained until retirement.
    pub fn prepare_copy(
        &self,
        destination: &Buffer,
        destination_offset: usize,
        source: &Buffer,
        source_offset: usize,
        length: usize,
    ) -> Result<PreparedGpu> {
        transfer_range(destination, destination_offset, length)?;
        transfer_range(source, source_offset, length)?;
        let dst = destination
            .device_address(self.device())?
            .checked_add(destination_offset as u64)
            .ok_or_else(|| Error::Message("copy address overflow".into()))?;
        let src = source
            .device_address(self.device())?
            .checked_add(source_offset as u64)
            .ok_or_else(|| Error::Message("copy address overflow".into()))?;
        if dst < src.saturating_add(length as u64) && src < dst.saturating_add(length as u64) {
            return Err(Error::Message("native copy ranges overlap".into()));
        }
        self.prepare_transfer(length, |offset, bytes| {
            self.prepare_copy_part(
                destination,
                destination_offset + offset,
                source,
                source_offset + offset,
                bytes,
            )
        })
    }

    fn prepare_transfer(
        &self,
        length: usize,
        mut prepare: impl FnMut(usize, usize) -> Result<PreparedGpu>,
    ) -> Result<PreparedGpu> {
        let sdma = self.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA;
        let limit = if sdma {
            sdma::transfer_limit(self.0.info.ring_byte_length as usize)?
        } else {
            // Preserve alignment at chunk boundaries and the transfer kernels'
            // bounded 32-bit dispatch extent after rounding to 256 work items.
            (u32::MAX as usize) & !255
        };
        if length <= limit {
            return prepare(0, length);
        }
        let mut parts = transfer_chunks(length, limit)
            .map(|(offset, bytes)| prepare(offset, bytes))
            .collect::<Result<Vec<_>>>()?;
        let mut last = parts.pop().expect("validated nonempty transfer");
        if sdma {
            last.prefix = Some(parts.into());
            Ok(last)
        } else {
            parts.push(last);
            // PM4 indirect batches fit the ring regardless of payload size.
            // Each part addresses disjoint bytes within the validated span.
            let commands: Vec<_> = parts.into_iter().map(|part| (part, false)).collect();
            unsafe { self.prepare_batch(&commands) }
        }
    }

    /// Prepare a byte copy between non-overlapping logical ranges.
    fn prepare_copy_part(
        &self,
        destination: &Buffer,
        destination_offset: usize,
        source: &Buffer,
        source_offset: usize,
        length: usize,
    ) -> Result<PreparedGpu> {
        transfer_range(destination, destination_offset, length)?;
        transfer_range(source, source_offset, length)?;
        if Arc::ptr_eq(&destination.0, &source.0)
            && destination_offset < source_offset + length
            && source_offset < destination_offset + length
        {
            return Err(Error::Message("native copy ranges overlap".into()));
        }
        if self.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA {
            return self.prepare_sdma_copy(
                destination,
                destination_offset,
                source,
                source_offset,
                length,
            );
        }
        let kernels = self.transfer_kernels()?;
        let wide = (destination_offset | source_offset | length) & 7 == 0;
        let copy = &kernels[if wide { 2 } else { 0 }];
        let elements = if wide { length / 8 } else { length };
        let count = (elements as u64).to_le_bytes();
        unsafe {
            self.prepare(
                copy,
                [elements.div_ceil(256) as u32, 1, 1],
                [256, 1, 1],
                &[
                    Argument::Buffer(destination, destination_offset),
                    Argument::Buffer(source, source_offset),
                    Argument::Value(&count),
                ],
            )
        }
    }
    pub(crate) fn prepare_copy_reserved(
        &self,
        destination: &Buffer,
        destination_offset: usize,
        source: &Buffer,
        source_offset: usize,
        length: usize,
        budget: Option<&crate::residency::MemoryBudget>,
    ) -> Result<PreparedGpu> {
        // Reserve before allocating native fences. One charge is shared by all
        // parts so it survives dropped wrappers and any accepted prefix.
        let count = if self.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA {
            length.div_ceil(sdma::transfer_limit(self.0.info.ring_byte_length as usize)?)
        } else {
            1
        };
        let bytes = count
            .checked_mul(64)
            .ok_or_else(|| Error::Message("transfer fence size overflow".into()))?;
        let reservation = budget
            .map(|budget| budget.reserve(bytes).map(Arc::new))
            .transpose()?;
        let mut command = self.prepare_copy(
            destination,
            destination_offset,
            source,
            source_offset,
            length,
        )?;
        if let Some(prefix) = &mut command.prefix {
            for part in Arc::get_mut(prefix).expect("new transfer prefix") {
                part.reserve_fence(reservation.clone());
            }
        }
        command.reserve_fence(reservation);
        Ok(command)
    }
    /// Prepare a byte fill, splitting large ranges for the selected engine.
    pub fn prepare_fill(
        &self,
        destination: &Buffer,
        offset: usize,
        length: usize,
        value: u8,
    ) -> Result<PreparedGpu> {
        transfer_range(destination, offset, length)?;
        self.prepare_transfer(length, |first, bytes| {
            self.prepare_fill_part(destination, offset + first, bytes, value)
        })
    }

    /// Prepare a repeated byte fill over one logical range.
    fn prepare_fill_part(
        &self,
        destination: &Buffer,
        offset: usize,
        length: usize,
        value: u8,
    ) -> Result<PreparedGpu> {
        transfer_range(destination, offset, length)?;
        if self.0.info.command_type == AMDF_QUEUE_COMMAND_TYPE_GPU_SDMA {
            return self.prepare_sdma_fill(destination, offset, length, value);
        }
        let kernels = self.transfer_kernels()?;
        let wide = (offset | length) & 7 == 0;
        let fill = &kernels[if wide { 3 } else { 1 }];
        let elements = if wide { length / 8 } else { length };
        let count = (elements as u64).to_le_bytes();
        let value = u32::from(value).to_le_bytes();
        unsafe {
            self.prepare(
                fill,
                [elements.div_ceil(256) as u32, 1, 1],
                [256, 1, 1],
                &[
                    Argument::Buffer(destination, offset),
                    Argument::Value(&count),
                    Argument::Value(&value),
                ],
            )
        }
    }
}
fn transfer_chunks(length: usize, limit: usize) -> impl Iterator<Item = (usize, usize)> {
    (0..length)
        .step_by(limit)
        .map(move |offset| (offset, (length - offset).min(limit)))
}

fn transfer_range(buffer: &Buffer, offset: usize, length: usize) -> Result<()> {
    if length == 0
        || offset
            .checked_add(length)
            .is_none_or(|end| end > buffer.len())
    {
        return Err(Error::Message(
            "native transfer range must be nonempty and in bounds".into(),
        ));
    }
    Ok(())
}

#[cfg(test)]
mod event_tests {
    use super::*;

    #[test]
    fn large_transfer_chunks_cover_boundaries_without_wrapping() {
        let limit = (u32::MAX as usize) & !255;
        for length in [
            1,
            limit - 1,
            limit,
            limit + 1,
            1usize << 32,
            (1usize << 33) + 19,
        ] {
            let mut end = 0;
            for (offset, bytes) in transfer_chunks(length, limit) {
                assert_eq!(offset, end);
                assert_eq!(offset % 8, 0);
                assert!((1..=limit).contains(&bytes));
                assert!(bytes.div_ceil(256) <= u32::MAX as usize / 256);
                end += bytes;
            }
            assert_eq!(end, length);
        }
    }

    #[test]
    #[ignore = "requires gfx1151 and native SDMA"]
    fn split_transfer_failure_drains_prefix_and_retains_fence_budget() -> Result<()> {
        let gpu = Device::open(Engine::Gpu, 0)?;
        let queue = gpu.sdma_queue()?;
        let limit = sdma::transfer_limit(queue.0.info.ring_byte_length as usize)?;
        let bytes = 2 * limit + 17;
        let source = gpu
            .fabric()
            .allocate_shared(bytes, std::slice::from_ref(&gpu))?;
        let target = gpu
            .fabric()
            .allocate_shared(bytes, std::slice::from_ref(&gpu))?;
        // Probe both completed chunks and the unsubmitted tail.
        for offset in [0, limit, 2 * limit] {
            source.write(offset, &[0x51])?;
        }
        target.write(2 * limit, &[0])?;
        let manager = crate::residency::ResidencyManager::new(4096)?;
        let command =
            queue.prepare_copy_reserved(&target, 0, &source, 0, bytes, Some(&manager.budget()))?;
        assert_eq!(command.parts().count(), 3);
        assert_eq!(manager.statistics().reserved_bytes, 3 * 64);
        command.exhaust_final_timeline();
        assert!(matches!(
            unsafe { command.dispatch() },
            Err(Error::DeviceLost(_))
        ));
        for offset in [0, limit] {
            let mut actual = [0];
            target.read(offset, &mut actual)?;
            assert_eq!(actual, [0x51]);
        }
        let mut tail = [0];
        target.read(2 * limit, &mut tail)?;
        assert_eq!(tail, [0]);
        for part in command.prefix.as_ref().unwrap().iter() {
            assert_eq!(part.inner.work.retired.load(Ordering::Acquire), 1);
        }
        drop(command);
        assert_eq!(manager.statistics().reserved_bytes, 0);
        Ok(())
    }

    #[test]
    fn a_full_ring_without_pending_work_returns_busy() {
        assert!(matches!(wait_for_capacity(None), Err(Error::Busy(_))));
    }

    #[test]
    fn stream_queue_configuration_is_positive_and_frozen_after_first_use() -> Result<()> {
        let mut pool = StreamQueues::default();
        assert!(pool.set_limit(0).is_err());
        pool.set_limit(2)?;
        assert_eq!(pool.limit, 2);
        assert!(
            pool.slots.is_empty(),
            "configuration must not allocate queues"
        );
        // Even an expired slot records that streams have used this device.
        pool.slots.push(Weak::new());
        pool.set_limit(2)?;
        assert!(matches!(pool.set_limit(8), Err(Error::Busy(_))));
        assert_eq!(pool.limit, 2);
        Ok(())
    }

    #[test]
    #[ignore = "requires gfx1151 and provisioned native libraries"]
    fn stream_queue_pool_is_bounded_shared_and_weak() -> Result<()> {
        let device = Device::open(Engine::Gpu, 0)?;
        device.set_stream_queue_count(2)?;
        let queues = std::thread::scope(|scope| {
            let workers: Vec<_> = (0..8)
                .map(|_| {
                    scope.spawn(|| {
                        let reopened = Device::open(Engine::Gpu, 0)?;
                        (0..16)
                            .map(|_| reopened.stream_queue())
                            .collect::<Result<Vec<_>>>()
                    })
                })
                .collect();
            workers
                .into_iter()
                .map(|worker| worker.join().unwrap())
                .collect::<Result<Vec<_>>>()
        })?;
        let identities: std::collections::HashSet<_> = queues
            .iter()
            .flatten()
            .map(|queue| Arc::as_ptr(&queue.0))
            .collect();
        assert_eq!(identities.len(), 2);
        let weak: Vec<_> = queues
            .iter()
            .flatten()
            .map(|queue| Arc::downgrade(&queue.0))
            .collect();
        drop(queues);
        assert!(weak.iter().all(|queue| queue.upgrade().is_none()));
        let replacement = device.stream_queue()?;
        assert_eq!(replacement.device().id(), device.id());
        Ok(())
    }

    #[test]
    #[ignore = "requires gfx1151 and provisioned native libraries"]
    fn shared_queue_submission_waits_for_other_producers() -> Result<()> {
        let device = Device::open(Engine::Gpu, 0)?;
        let fabric = device.fabric();
        let queue = device.stream_queue()?;
        let gate = Completion {
            queue: queue.clone(),
            work: Arc::new(Work {
                fence: fabric.allocate_shared(64, std::slice::from_ref(&device))?,
                _kernels: Vec::new(),
                buffers: Vec::new(),
                leases: Mutex::new(Vec::new()),
                retired: AtomicU32::new(0),
                active: AtomicU32::new(1),
                dependencies: Vec::new(),
            }),
            value: 1,
        };
        let buffer = fabric.allocate(64, std::slice::from_ref(&device))?;
        let fill = queue.prepare_fill(&buffer, 0, 64, 19)?;
        let final_fill = queue.prepare_fill(&buffer, 0, 64, 42)?;
        unsafe { queue.prepare_wait(&gate)?.dispatch()? };
        let mut submitted = 0;
        loop {
            match unsafe { fill.dispatch() } {
                Ok(_) => submitted += 1,
                Err(Error::Busy(_)) => break,
                Err(error) => {
                    gate.work.fence.write(0, &1u32.to_le_bytes())?;
                    return Err(error);
                }
            }
        }
        let (sent, received) = std::sync::mpsc::channel();
        let worker = std::thread::spawn(move || {
            sent.send(unsafe { final_fill.dispatch_wait() }).unwrap();
        });
        let blocked = received.recv_timeout(Duration::from_millis(50));
        gate.work.fence.write(0, &1u32.to_le_bytes())?;
        worker.join().unwrap();
        assert!(submitted > 0);
        assert!(matches!(
            blocked,
            Err(std::sync::mpsc::RecvTimeoutError::Timeout)
        ));
        let done = received.recv().unwrap()?;
        assert!(done.wait_timeout(Duration::from_secs(5))?);
        let mut actual = [0; 64];
        buffer.read(0, &mut actual)?;
        assert_eq!(actual, [42; 64]);
        Ok(())
    }

    #[test]
    fn graph_batch_ranges_reserve_barriers_markers_and_padding() -> Result<()> {
        assert!(batch_ranges([], false)?.is_empty());
        assert_eq!(batch_ranges([(100, false); 3], false)?, vec![0..3]);
        let near_limit = INDIRECT_WORD_LIMIT - 20;
        assert_eq!(
            batch_ranges([(near_limit, true), (1, false)], false)?,
            vec![0..1, 1..2]
        );
        assert!(batch_ranges([(INDIRECT_WORD_LIMIT - 9, false)], false).is_err());
        assert!(batch_ranges([(usize::MAX, true)], true).is_err());
        assert_eq!(
            batch_ranges([(INDIRECT_WORD_LIMIT / 2 - 10, false); 2], false)?,
            vec![0..2]
        );
        assert_eq!(
            batch_ranges([(INDIRECT_WORD_LIMIT / 2 - 10, false); 2], true)?,
            vec![0..1, 1..2]
        );
        Ok(())
    }

    #[test]
    #[ignore = "requires gfx1151 and provisioned native libraries"]
    fn observing_an_event_retires_every_earlier_submission() -> Result<()> {
        let device = Device::open(Engine::Gpu, 0)?;
        let fabric = device.fabric();
        let producer = device.queue()?;
        let consumer = device.queue()?;
        // A host-controlled fence keeps the whole producer prefix outstanding
        // while the consumer snapshots its event. No timing race or slow kernel
        // is needed to keep the earlier buffer's lease alive.
        let gate = Completion {
            queue: device.queue()?,
            work: Arc::new(Work {
                fence: fabric.allocate_shared(64, std::slice::from_ref(&device))?,
                _kernels: Vec::new(),
                buffers: Vec::new(),
                leases: Mutex::new(Vec::new()),
                retired: AtomicU32::new(0),
                active: AtomicU32::new(1),
                dependencies: Vec::new(),
            }),
            value: 1,
        };
        let first = fabric.allocate(64, std::slice::from_ref(&device))?;
        let last = fabric.allocate(64, std::slice::from_ref(&device))?;
        unsafe { producer.prepare_wait(&gate)?.dispatch()? };
        unsafe {
            producer
                .prepare_fill(&first, 0, first.len(), 37)?
                .dispatch()?
        };
        let event = unsafe {
            producer
                .prepare_fill(&last, 0, last.len(), 91)?
                .dispatch()?
        };
        let observed = unsafe { consumer.prepare_wait(&event)?.dispatch()? };
        assert!(matches!(first.read(0, &mut [0; 64]), Err(Error::Busy(_))));
        gate.work.fence.write(0, &1u32.to_le_bytes())?;
        if !observed.wait_timeout(Duration::from_secs(5))? {
            std::mem::forget(observed);
            return Err(Error::DeviceLost("event prefix did not retire".into()));
        }
        // No producer wait or poll: consumer completion must retire all resources
        // ordered by the event, including a buffer absent from its last command.
        let mut bytes = [0; 64];
        first.read(0, &mut bytes)?;
        assert_eq!(bytes, [37; 64]);
        last.read(0, &mut bytes)?;
        assert_eq!(bytes, [91; 64]);
        Ok(())
    }
}