Struct Stream 

pub struct Stream { /* private fields */ }

An ordered command stream. Mutating operations require exclusive access. Prepared kernels and allocations can be reused without a compiler/cache lookup.

Implementations

impl Stream

pub fn access_graph(&self) -> Result<AccessGraph<'_>>

Begin an access-aware graph recording on this stream.

impl Stream

pub fn with_memory_budget(self, budget: MemoryBudget) -> Self

Charge subsequent allocations (including native upload/readback staging) against a shared residency ceiling. Existing allocations are unchanged. Charges remain with buffers, queued uses and recorded graphs, and leak with quarantined native storage after an uncertain failure. The ceiling covers requested extents, not native allocator rounding or code objects.

pub fn memory_budget(&self) -> Option<&MemoryBudget>

Allocation ceiling for this stream, including transfer staging.

pub fn target(&self) -> &Target

Architecture reported by this stream’s device.

pub fn record_event(&mut self) -> Result<Event>

Submit preceding work and record a single immutable completion event. Recording does not wait on the host and does not reclaim upload staging.

pub fn wait_event(&mut self, event: &Event) -> Result<()>

Queue a device-side dependency before subsequent work on this stream. The event must come from the same device. This call does not wait on the host.

pub fn open() -> Result<Self>

Open an ordered stream on device zero.

Examples found in repository

examples/graph_prepare.rs (line 7)

fn main() -> hrx::Result<()> {
    let runtime = Runtime::new()?;
    let buffer = runtime.allocate(4096, MemoryPlacement::GpuLocal)?;
    let stream = Stream::open()?;
    let raw = stream.allocate(4096)?;
    for count in [128, 512, 1024, 2048] {
        let begin = Instant::now();
        let mut graph = runtime.graph();
        for _ in 0..count {
            graph.fill(buffer.view(), 0)?;
        }
        let graph = graph.prepare()?;
        let inferred = begin.elapsed();
        let begin = Instant::now();
        let mut native = stream.graph()?;
        let mut last = None;
        for _ in 0..count {
            last = Some(native.fill(last.as_slice(), raw.binding(), 0)?);
        }
        let _native = native.finish()?;
        println!(
            "nodes={count} inferred_us={} explicit_us={}",
            inferred.as_micros(),
            begin.elapsed().as_micros()
        );
        graph.submit()?.wait()?;
    }
    Ok(())
}

examples/gemm_pipeline.rs (line 36)

fn gpu_kernel(
    runtime: &Runtime,
    source: &str,
    symbol: &str,
    config: &[(&str, String)],
    elements: usize,
    bindings: Vec<BindingContract>,
) -> Result<GpuKernel> {
    let compiler = hrx::loom::Compiler::for_target(None, runtime.gpu()?.target())?;
    let module = compiler.module(source);
    let mut spec = hrx::loom::Specialization::new(symbol);
    for (key, value) in config {
        spec.set_config(*key, value.clone());
    }
    let artifact = module.compile(&spec)?;
    let stream = hrx::gpu::Stream::open()?;
    let raw = unsafe { stream.load_artifact(&artifact) }?;
    let mut constants = hrx::gpu::Constants::new();
    match raw.info().constant_byte_length {
        8 => constants.push(elements as u32)?,
        12 => constants.push(elements as u64)?,
        _ => return Err(hrx::Error::Message("unexpected scalar ABI".into())),
    }
    constants.push(1f32)?;
    unsafe {
        runtime.load_gpu_kernel(
            artifact.path(),
            symbol,
            [elements.div_ceil(256) as u32, 1, 1],
            [256, 1, 1],
            KernelContract {
                bindings,
                constants: constants.as_bytes().to_vec(),
            },
        )
    }
}

examples/import_probe.rs (line 32)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/gpu_load.rs (line 97)

fn main() -> Result<()> {
    let options = match parse() {
        Ok(options) => options,
        Err(message) => {
            eprintln!("gpu_load: {message}");
            std::process::exit(2);
        }
    };
    let size = options.mib * 1024 * 1024;
    let mut stream = Stream::open()?;
    let source = stream.allocate(size)?;
    let destination = stream.allocate(size)?;
    // Touch both allocations so the steady-state loop never pays first-use faults.
    stream.fill(source.binding(), 0x3c)?;
    stream.fill(destination.binding(), 0)?;
    stream.synchronize()?;

    let run_batch = |stream: &mut Stream| -> Result<()> {
        for _ in 0..options.batch {
            match options.mode {
                Mode::Copy => stream.copy(destination.binding(), source.binding())?,
                Mode::Fill => stream.fill(destination.binding(), 0x5a)?,
            }
        }
        stream.synchronize()
    };

    // Warm up outside the measured window: the first batch pays queue and paging costs.
    run_batch(&mut stream)?;

    let window = Duration::from_secs_f64(options.seconds);
    let start = Instant::now();
    let mut iterations = 0usize;
    // Per-batch rates, so the harness can see whether contention was steady or bursty.
    let mut batch_rates = Vec::new();
    while start.elapsed() < window {
        let batch_start = Instant::now();
        run_batch(&mut stream)?;
        let seconds = batch_start.elapsed().as_secs_f64();
        iterations += options.batch;
        let bytes = options.mode.traffic(size) * options.batch;
        batch_rates.push(bytes as f64 / seconds / 1e9);
    }
    let elapsed = start.elapsed().as_secs_f64();
    let total = options.mode.traffic(size) as f64 * iterations as f64;
    let mean = total / elapsed / 1e9;
    batch_rates.sort_by(f64::total_cmp);
    let median = batch_rates[batch_rates.len() / 2];
    let low = batch_rates[batch_rates.len() / 20];

    println!(
        concat!(
            r#"{{"device":"gpu","mode":"{}","buffer_mib":{},"batch":{},"#,
            r#""seconds":{:.4},"iterations":{},"gb_s_mean":{:.3},"#,
            r#""gb_s_median":{:.3},"gb_s_p5":{:.3}}}"#
        ),
        options.mode.name(),
        options.mib,
        options.batch,
        elapsed,
        iterations,
        mean,
        median,
        low
    );
    eprintln!(
        "gpu_load: {} {:.1} GB/s mean over {:.1}s ({} iterations)",
        options.mode.name(),
        mean,
        elapsed,
        iterations
    );
    Ok(())
}

examples/unified_probe.rs (line 127)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/stream_bench.rs (line 31)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

Additional examples can be found in:

• examples/dmabuf_probe.rs

pub fn device_id(&self) -> usize

Identifies the device this stream runs on, for callers that keep per-device state. Kernels are device-scoped, so a cache of loaded executables is only valid for the device that loaded them.

pub fn id(&self) -> usize

Identifies this stream, for callers that keep per-stream state.

Buffers are device-scoped, so nothing here rejects one used on a sibling stream — correct, because events can order that. What events cannot fix is reuse: a pool handing a buffer out again relies on the queue that used it last running in order, which holds within a stream and not across them. A caller pooling allocations needs to say which stream a block came from, and this is how. Unique among live streams; a value may repeat once its stream is dropped.

pub fn allocate(&self, bytes: usize) -> Result<Buffer>

Allocate storage owned by this stream; zero bytes is rounded to one.

Examples found in repository

examples/graph_prepare.rs (line 8)

fn main() -> hrx::Result<()> {
    let runtime = Runtime::new()?;
    let buffer = runtime.allocate(4096, MemoryPlacement::GpuLocal)?;
    let stream = Stream::open()?;
    let raw = stream.allocate(4096)?;
    for count in [128, 512, 1024, 2048] {
        let begin = Instant::now();
        let mut graph = runtime.graph();
        for _ in 0..count {
            graph.fill(buffer.view(), 0)?;
        }
        let graph = graph.prepare()?;
        let inferred = begin.elapsed();
        let begin = Instant::now();
        let mut native = stream.graph()?;
        let mut last = None;
        for _ in 0..count {
            last = Some(native.fill(last.as_slice(), raw.binding(), 0)?);
        }
        let _native = native.finish()?;
        println!(
            "nodes={count} inferred_us={} explicit_us={}",
            inferred.as_micros(),
            begin.elapsed().as_micros()
        );
        graph.submit()?.wait()?;
    }
    Ok(())
}

examples/import_probe.rs (line 59)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/gpu_load.rs (line 98)

fn main() -> Result<()> {
    let options = match parse() {
        Ok(options) => options,
        Err(message) => {
            eprintln!("gpu_load: {message}");
            std::process::exit(2);
        }
    };
    let size = options.mib * 1024 * 1024;
    let mut stream = Stream::open()?;
    let source = stream.allocate(size)?;
    let destination = stream.allocate(size)?;
    // Touch both allocations so the steady-state loop never pays first-use faults.
    stream.fill(source.binding(), 0x3c)?;
    stream.fill(destination.binding(), 0)?;
    stream.synchronize()?;

    let run_batch = |stream: &mut Stream| -> Result<()> {
        for _ in 0..options.batch {
            match options.mode {
                Mode::Copy => stream.copy(destination.binding(), source.binding())?,
                Mode::Fill => stream.fill(destination.binding(), 0x5a)?,
            }
        }
        stream.synchronize()
    };

    // Warm up outside the measured window: the first batch pays queue and paging costs.
    run_batch(&mut stream)?;

    let window = Duration::from_secs_f64(options.seconds);
    let start = Instant::now();
    let mut iterations = 0usize;
    // Per-batch rates, so the harness can see whether contention was steady or bursty.
    let mut batch_rates = Vec::new();
    while start.elapsed() < window {
        let batch_start = Instant::now();
        run_batch(&mut stream)?;
        let seconds = batch_start.elapsed().as_secs_f64();
        iterations += options.batch;
        let bytes = options.mode.traffic(size) * options.batch;
        batch_rates.push(bytes as f64 / seconds / 1e9);
    }
    let elapsed = start.elapsed().as_secs_f64();
    let total = options.mode.traffic(size) as f64 * iterations as f64;
    let mean = total / elapsed / 1e9;
    batch_rates.sort_by(f64::total_cmp);
    let median = batch_rates[batch_rates.len() / 2];
    let low = batch_rates[batch_rates.len() / 20];

    println!(
        concat!(
            r#"{{"device":"gpu","mode":"{}","buffer_mib":{},"batch":{},"#,
            r#""seconds":{:.4},"iterations":{},"gb_s_mean":{:.3},"#,
            r#""gb_s_median":{:.3},"gb_s_p5":{:.3}}}"#
        ),
        options.mode.name(),
        options.mib,
        options.batch,
        elapsed,
        iterations,
        mean,
        median,
        low
    );
    eprintln!(
        "gpu_load: {} {:.1} GB/s mean over {:.1}s ({} iterations)",
        options.mode.name(),
        mean,
        elapsed,
        iterations
    );
    Ok(())
}

examples/unified_probe.rs (line 199)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/stream_bench.rs (line 49)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn allocate_zeroed(&self, bytes: usize) -> Result<Buffer>

Allocate device-local storage and enqueue a zero fill over its requested size.

pub unsafe fn import_host( &self, pointer: *mut c_void, bytes: usize, ) -> Result<Buffer>

Import host memory the caller owns as a device-visible buffer, without copying.

On an APU the GPU and the NPU address the same physical pages, so importing one host allocation into both runtimes is the basis of zero-copy handoff between them: the same pages can simultaneously back an XRT BO driving the NPU.

Safety

pointer must be page-aligned, cover at least bytes, and stay allocated and unmoved for the whole life of the returned buffer – the buffer borrows the memory and never frees it. The host must not read or write those bytes while device work touching them is in flight.

Examples found in repository

examples/import_probe.rs (line 33)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/unified_probe.rs (line 128)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

pub fn synchronize(&mut self) -> Result<()>

Submit and wait for all work, then reclaim completed upload staging.

Examples found in repository

examples/stream_bench.rs (line 14)

fn measure(
    stream: &mut Stream,
    samples: usize,
    operations: usize,
    mut run: impl FnMut(&mut Stream) -> Result<()>,
) -> Result<(f64, f64)> {
    let mut enqueue = Vec::new();
    let mut complete = Vec::new();
    for sample in 0..samples + 3 {
        stream.synchronize()?;
        let start = Instant::now();
        run(stream)?;
        let recorded = start.elapsed();
        stream.synchronize()?;
        let finished = start.elapsed();
        if sample >= 3 {
            enqueue.push(recorded.as_secs_f64() * 1e9 / operations as f64);
            complete.push(finished.as_secs_f64() * 1e9 / operations as f64);
        }
    }
    enqueue.sort_by(f64::total_cmp);
    complete.sort_by(f64::total_cmp);
    Ok((enqueue[samples / 2], complete[samples / 2]))
}

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

examples/import_probe.rs (line 47)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/gpu_load.rs (line 103)

fn main() -> Result<()> {
    let options = match parse() {
        Ok(options) => options,
        Err(message) => {
            eprintln!("gpu_load: {message}");
            std::process::exit(2);
        }
    };
    let size = options.mib * 1024 * 1024;
    let mut stream = Stream::open()?;
    let source = stream.allocate(size)?;
    let destination = stream.allocate(size)?;
    // Touch both allocations so the steady-state loop never pays first-use faults.
    stream.fill(source.binding(), 0x3c)?;
    stream.fill(destination.binding(), 0)?;
    stream.synchronize()?;

    let run_batch = |stream: &mut Stream| -> Result<()> {
        for _ in 0..options.batch {
            match options.mode {
                Mode::Copy => stream.copy(destination.binding(), source.binding())?,
                Mode::Fill => stream.fill(destination.binding(), 0x5a)?,
            }
        }
        stream.synchronize()
    };

    // Warm up outside the measured window: the first batch pays queue and paging costs.
    run_batch(&mut stream)?;

    let window = Duration::from_secs_f64(options.seconds);
    let start = Instant::now();
    let mut iterations = 0usize;
    // Per-batch rates, so the harness can see whether contention was steady or bursty.
    let mut batch_rates = Vec::new();
    while start.elapsed() < window {
        let batch_start = Instant::now();
        run_batch(&mut stream)?;
        let seconds = batch_start.elapsed().as_secs_f64();
        iterations += options.batch;
        let bytes = options.mode.traffic(size) * options.batch;
        batch_rates.push(bytes as f64 / seconds / 1e9);
    }
    let elapsed = start.elapsed().as_secs_f64();
    let total = options.mode.traffic(size) as f64 * iterations as f64;
    let mean = total / elapsed / 1e9;
    batch_rates.sort_by(f64::total_cmp);
    let median = batch_rates[batch_rates.len() / 2];
    let low = batch_rates[batch_rates.len() / 20];

    println!(
        concat!(
            r#"{{"device":"gpu","mode":"{}","buffer_mib":{},"batch":{},"#,
            r#""seconds":{:.4},"iterations":{},"gb_s_mean":{:.3},"#,
            r#""gb_s_median":{:.3},"gb_s_p5":{:.3}}}"#
        ),
        options.mode.name(),
        options.mib,
        options.batch,
        elapsed,
        iterations,
        mean,
        median,
        low
    );
    eprintln!(
        "gpu_load: {} {:.1} GB/s mean over {:.1}s ({} iterations)",
        options.mode.name(),
        mean,
        elapsed,
        iterations
    );
    Ok(())
}

examples/unified_probe.rs (line 137)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/dmabuf_probe.rs (line 195)

fn main() -> Result<(), String> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().map_err(|_| "--mib expects an integer")?,
        ),
        _ => return Err("usage: dmabuf_probe <xclbin> [--mib N]".into()),
    };
    let bytes = mib * 1024 * 1024;
    let hsa = Hsa::open()?;

    unsafe {
        let init: libloading::Symbol<unsafe extern "C" fn() -> Status> =
            hsa.symbol(b"hsa_init\0")?;
        let status = init();
        if status != HSA_STATUS_SUCCESS {
            return Err(format!("hsa_init returned {status}"));
        }

        // --- find a GPU agent -------------------------------------------------
        let iterate_agents: libloading::Symbol<
            unsafe extern "C" fn(
                unsafe extern "C" fn(u64, *mut c_void) -> Status,
                *mut c_void,
            ) -> Status,
        > = hsa.symbol(b"hsa_iterate_agents\0")?;
        let agent_info: libloading::Symbol<unsafe extern "C" fn(u64, u32, *mut c_void) -> Status> =
            hsa.symbol(b"hsa_agent_get_info\0")?;

        // The callback needs the symbol; pass it through the user-data pointer.
        struct AgentScan<'a> {
            info: &'a dyn Fn(u64, u32, *mut c_void) -> Status,
            found: Found,
        }
        unsafe extern "C" fn on_agent(agent: u64, data: *mut c_void) -> Status {
            let scan = unsafe { &mut *data.cast::<AgentScan>() };
            let mut kind: u32 = 0;
            if (scan.info)(agent, HSA_AGENT_INFO_DEVICE, (&raw mut kind).cast())
                == HSA_STATUS_SUCCESS
                && kind == HSA_DEVICE_TYPE_GPU
                && scan.found.agent == 0
            {
                scan.found.agent = agent;
            }
            HSA_STATUS_SUCCESS
        }
        let info_fn = |agent, key, out| agent_info(agent, key, out);
        let mut scan = AgentScan {
            info: &info_fn,
            found: Found::default(),
        };
        iterate_agents(on_agent, (&raw mut scan).cast());
        if scan.found.agent == 0 {
            return Err("no GPU agent found".into());
        }
        println!("gpu agent: 0x{:x}", scan.found.agent);

        // --- find an allocatable global pool ----------------------------------
        let iterate_pools: libloading::Symbol<
            unsafe extern "C" fn(
                u64,
                unsafe extern "C" fn(u64, *mut c_void) -> Status,
                *mut c_void,
            ) -> Status,
        > = hsa.symbol(b"hsa_amd_agent_iterate_memory_pools\0")?;
        let pool_info: libloading::Symbol<unsafe extern "C" fn(u64, u32, *mut c_void) -> Status> =
            hsa.symbol(b"hsa_amd_memory_pool_get_info\0")?;

        struct PoolScan<'a> {
            info: &'a dyn Fn(u64, u32, *mut c_void) -> Status,
            found: Found,
        }
        unsafe extern "C" fn on_pool(pool: u64, data: *mut c_void) -> Status {
            let scan = unsafe { &mut *data.cast::<PoolScan>() };
            if scan.found.pool != 0 {
                return HSA_STATUS_SUCCESS;
            }
            let mut segment: u32 = u32::MAX;
            let mut allowed: bool = false;
            let mut size: usize = 0;
            (scan.info)(pool, POOL_INFO_SEGMENT, (&raw mut segment).cast());
            (scan.info)(pool, POOL_INFO_ALLOC_ALLOWED, (&raw mut allowed).cast());
            (scan.info)(pool, POOL_INFO_SIZE, (&raw mut size).cast());
            if segment == SEGMENT_GLOBAL && allowed && size > 0 {
                scan.found.pool = pool;
                scan.found.pool_bytes = size;
            }
            HSA_STATUS_SUCCESS
        }
        let pool_fn = |pool, key, out| pool_info(pool, key, out);
        let mut pools = PoolScan {
            info: &pool_fn,
            found: Found::default(),
        };
        iterate_pools(scan.found.agent, on_pool, (&raw mut pools).cast());
        if pools.found.pool == 0 {
            return Err("no allocatable global memory pool on the GPU agent".into());
        }
        println!(
            "gpu memory pool: 0x{:x} ({} MiB)",
            pools.found.pool,
            pools.found.pool_bytes / (1024 * 1024)
        );

        // --- allocate from it and export a dma-buf ----------------------------
        let allocate: libloading::Symbol<
            unsafe extern "C" fn(u64, usize, u32, *mut *mut c_void) -> Status,
        > = hsa.symbol(b"hsa_amd_memory_pool_allocate\0")?;
        let mut pointer: *mut c_void = std::ptr::null_mut();
        let status = allocate(pools.found.pool, bytes, 0, &raw mut pointer);
        if status != HSA_STATUS_SUCCESS {
            return Err(format!("hsa_amd_memory_pool_allocate returned {status}"));
        }
        println!("pool allocation: {mib} MiB at {pointer:p}");

        // The route that would make a shared buffer a first-class hrx buffer: allocate it
        // through the GPU runtime, take its device pointer, and export that.
        match hrx::Stream::open() {
            Ok(mut stream) => match stream.allocate_shared(bytes) {
                Ok(buffer) => {
                    stream.fill(buffer.binding(), 0x11).ok();
                    stream.synchronize().ok();
                    match buffer.device_ptr() {
                        Ok(hrx_pointer) => {
                            println!("hrx shared allocation: ok, device ptr {hrx_pointer:p}");
                            let mut fd: i32 = -1;
                            let mut offset: u64 = 0;
                            let export: libloading::Symbol<
                                unsafe extern "C" fn(
                                    *const c_void,
                                    usize,
                                    *mut i32,
                                    *mut u64,
                                ) -> Status,
                            > = hsa.symbol(b"hsa_amd_portable_export_dmabuf\0")?;
                            let status = export(hrx_pointer, bytes, &raw mut fd, &raw mut offset);
                            if status != HSA_STATUS_SUCCESS {
                                println!("  dma-buf export of an hrx buffer: NO (status {status})");
                            } else {
                                println!(
                                    "  dma-buf export of an hrx buffer: YES (fd {fd}, offset {offset})"
                                );
                                match dvxrt::Context::new(0, &xclbin) {
                                    Ok(context) => match context.import_dmabuf(fd, bytes) {
                                        Ok(bo) => {
                                            let mut seen = vec![0u8; 4096];
                                            bo.read(&mut seen).ok();
                                            let wrong = seen.iter().filter(|&&b| b != 0x11).count();
                                            println!(
                                                "  npu reads the hrx buffer: {}",
                                                if wrong == 0 {
                                                    "YES -- one buffer, both engines".into()
                                                } else {
                                                    format!("NO ({wrong}/4096 differ)")
                                                }
                                            );
                                        }
                                        Err(error) => println!("  npu import: NO ({error})"),
                                    },
                                    Err(error) => println!("  npu context: {error}"),
                                }
                            }
                        }
                        Err(error) => println!("hrx shared allocation device ptr: NO ({error})"),
                    }
                }
                Err(error) => println!("hrx shared allocation: NO ({error})"),
            },
            Err(error) => println!("hrx stream unavailable: {error}"),
        }

        let export: libloading::Symbol<
            unsafe extern "C" fn(*const c_void, usize, *mut i32, *mut u64) -> Status,
        > = hsa.symbol(b"hsa_amd_portable_export_dmabuf\0")?;
        let mut fd: i32 = -1;
        let mut offset: u64 = 0;
        let status = export(pointer, bytes, &raw mut fd, &raw mut offset);
        if status != HSA_STATUS_SUCCESS {
            println!("hsa dma-buf export of pool memory: NO (status {status})");
            return Ok(());
        }
        println!("hsa dma-buf export of pool memory: YES (fd {fd}, offset {offset})");

        // --- and hand that descriptor to the NPU ------------------------------
        match dvxrt::Context::new(0, &xclbin) {
            Ok(context) => {
                let _group = context.group_id(ARG_A).map_err(|e| e.to_string())?;
                match context.import_dmabuf(fd, bytes) {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        // Import alone only proves the descriptor was accepted. Write with
                        // the GPU and read through the NPU's BO to prove it is one buffer.
                        let fill: libloading::Symbol<
                            unsafe extern "C" fn(*mut c_void, u32, usize) -> Status,
                        > = hsa.symbol(b"hsa_amd_memory_fill\0")?;
                        let pattern = 0xa5a5_a5a5u32;
                        let status = fill(pointer, pattern, bytes / 4);
                        if status != HSA_STATUS_SUCCESS {
                            println!("  gpu fill of pool memory failed (status {status})");
                            return Ok(());
                        }
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != 0xa5).count();
                                if wrong == 0 {
                                    println!(
                                        "  gpu write -> npu read over dma-buf: YES -- zero-copy confirmed"
                                    );
                                } else {
                                    println!(
                                        "  gpu write -> npu read over dma-buf: NO ({wrong}/4096 bytes differ)"
                                    );
                                }
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("npu context: UNAVAILABLE ({error})"),
        }
    }
    Ok(())
}

pub fn upload_blocking(&mut self, dst: View<'_>, bytes: &[u8]) -> Result<()>

Drain pending work, then upload bytes at the start of a view and wait for completion. The input must fit within the view. Completed staging is reclaimed.

Examples found in repository

examples/stream_bench.rs (line 87)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn upload_blocking_at( &mut self, dst: &Buffer, offset: usize, bytes: &[u8], ) -> Result<()>

Synchronously upload at a checked byte offset into an allocation.

pub fn read_blocking(&mut self, src: View<'_>, bytes: &mut [u8]) -> Result<()>

Drain all pending work before a synchronous read and reclaim staging.

Examples found in repository

examples/import_probe.rs (line 63)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/unified_probe.rs (line 202)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/stream_bench.rs (lines 69-72)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn read_blocking_at( &mut self, src: &Buffer, offset: usize, bytes: &mut [u8], ) -> Result<()>

Synchronously read at a checked byte offset from an allocation.

pub fn fill(&self, dst: View<'_>, value: u8) -> Result<()>

Queue a byte-pattern fill over a nonempty view owned by this stream.

Examples found in repository

examples/import_probe.rs (line 46)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/gpu_load.rs (line 101)

fn main() -> Result<()> {
    let options = match parse() {
        Ok(options) => options,
        Err(message) => {
            eprintln!("gpu_load: {message}");
            std::process::exit(2);
        }
    };
    let size = options.mib * 1024 * 1024;
    let mut stream = Stream::open()?;
    let source = stream.allocate(size)?;
    let destination = stream.allocate(size)?;
    // Touch both allocations so the steady-state loop never pays first-use faults.
    stream.fill(source.binding(), 0x3c)?;
    stream.fill(destination.binding(), 0)?;
    stream.synchronize()?;

    let run_batch = |stream: &mut Stream| -> Result<()> {
        for _ in 0..options.batch {
            match options.mode {
                Mode::Copy => stream.copy(destination.binding(), source.binding())?,
                Mode::Fill => stream.fill(destination.binding(), 0x5a)?,
            }
        }
        stream.synchronize()
    };

    // Warm up outside the measured window: the first batch pays queue and paging costs.
    run_batch(&mut stream)?;

    let window = Duration::from_secs_f64(options.seconds);
    let start = Instant::now();
    let mut iterations = 0usize;
    // Per-batch rates, so the harness can see whether contention was steady or bursty.
    let mut batch_rates = Vec::new();
    while start.elapsed() < window {
        let batch_start = Instant::now();
        run_batch(&mut stream)?;
        let seconds = batch_start.elapsed().as_secs_f64();
        iterations += options.batch;
        let bytes = options.mode.traffic(size) * options.batch;
        batch_rates.push(bytes as f64 / seconds / 1e9);
    }
    let elapsed = start.elapsed().as_secs_f64();
    let total = options.mode.traffic(size) as f64 * iterations as f64;
    let mean = total / elapsed / 1e9;
    batch_rates.sort_by(f64::total_cmp);
    let median = batch_rates[batch_rates.len() / 2];
    let low = batch_rates[batch_rates.len() / 20];

    println!(
        concat!(
            r#"{{"device":"gpu","mode":"{}","buffer_mib":{},"batch":{},"#,
            r#""seconds":{:.4},"iterations":{},"gb_s_mean":{:.3},"#,
            r#""gb_s_median":{:.3},"gb_s_p5":{:.3}}}"#
        ),
        options.mode.name(),
        options.mib,
        options.batch,
        elapsed,
        iterations,
        mean,
        median,
        low
    );
    eprintln!(
        "gpu_load: {} {:.1} GB/s mean over {:.1}s ({} iterations)",
        options.mode.name(),
        mean,
        elapsed,
        iterations
    );
    Ok(())
}

examples/unified_probe.rs (line 136)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/dmabuf_probe.rs (line 194)

fn main() -> Result<(), String> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().map_err(|_| "--mib expects an integer")?,
        ),
        _ => return Err("usage: dmabuf_probe <xclbin> [--mib N]".into()),
    };
    let bytes = mib * 1024 * 1024;
    let hsa = Hsa::open()?;

    unsafe {
        let init: libloading::Symbol<unsafe extern "C" fn() -> Status> =
            hsa.symbol(b"hsa_init\0")?;
        let status = init();
        if status != HSA_STATUS_SUCCESS {
            return Err(format!("hsa_init returned {status}"));
        }

        // --- find a GPU agent -------------------------------------------------
        let iterate_agents: libloading::Symbol<
            unsafe extern "C" fn(
                unsafe extern "C" fn(u64, *mut c_void) -> Status,
                *mut c_void,
            ) -> Status,
        > = hsa.symbol(b"hsa_iterate_agents\0")?;
        let agent_info: libloading::Symbol<unsafe extern "C" fn(u64, u32, *mut c_void) -> Status> =
            hsa.symbol(b"hsa_agent_get_info\0")?;

        // The callback needs the symbol; pass it through the user-data pointer.
        struct AgentScan<'a> {
            info: &'a dyn Fn(u64, u32, *mut c_void) -> Status,
            found: Found,
        }
        unsafe extern "C" fn on_agent(agent: u64, data: *mut c_void) -> Status {
            let scan = unsafe { &mut *data.cast::<AgentScan>() };
            let mut kind: u32 = 0;
            if (scan.info)(agent, HSA_AGENT_INFO_DEVICE, (&raw mut kind).cast())
                == HSA_STATUS_SUCCESS
                && kind == HSA_DEVICE_TYPE_GPU
                && scan.found.agent == 0
            {
                scan.found.agent = agent;
            }
            HSA_STATUS_SUCCESS
        }
        let info_fn = |agent, key, out| agent_info(agent, key, out);
        let mut scan = AgentScan {
            info: &info_fn,
            found: Found::default(),
        };
        iterate_agents(on_agent, (&raw mut scan).cast());
        if scan.found.agent == 0 {
            return Err("no GPU agent found".into());
        }
        println!("gpu agent: 0x{:x}", scan.found.agent);

        // --- find an allocatable global pool ----------------------------------
        let iterate_pools: libloading::Symbol<
            unsafe extern "C" fn(
                u64,
                unsafe extern "C" fn(u64, *mut c_void) -> Status,
                *mut c_void,
            ) -> Status,
        > = hsa.symbol(b"hsa_amd_agent_iterate_memory_pools\0")?;
        let pool_info: libloading::Symbol<unsafe extern "C" fn(u64, u32, *mut c_void) -> Status> =
            hsa.symbol(b"hsa_amd_memory_pool_get_info\0")?;

        struct PoolScan<'a> {
            info: &'a dyn Fn(u64, u32, *mut c_void) -> Status,
            found: Found,
        }
        unsafe extern "C" fn on_pool(pool: u64, data: *mut c_void) -> Status {
            let scan = unsafe { &mut *data.cast::<PoolScan>() };
            if scan.found.pool != 0 {
                return HSA_STATUS_SUCCESS;
            }
            let mut segment: u32 = u32::MAX;
            let mut allowed: bool = false;
            let mut size: usize = 0;
            (scan.info)(pool, POOL_INFO_SEGMENT, (&raw mut segment).cast());
            (scan.info)(pool, POOL_INFO_ALLOC_ALLOWED, (&raw mut allowed).cast());
            (scan.info)(pool, POOL_INFO_SIZE, (&raw mut size).cast());
            if segment == SEGMENT_GLOBAL && allowed && size > 0 {
                scan.found.pool = pool;
                scan.found.pool_bytes = size;
            }
            HSA_STATUS_SUCCESS
        }
        let pool_fn = |pool, key, out| pool_info(pool, key, out);
        let mut pools = PoolScan {
            info: &pool_fn,
            found: Found::default(),
        };
        iterate_pools(scan.found.agent, on_pool, (&raw mut pools).cast());
        if pools.found.pool == 0 {
            return Err("no allocatable global memory pool on the GPU agent".into());
        }
        println!(
            "gpu memory pool: 0x{:x} ({} MiB)",
            pools.found.pool,
            pools.found.pool_bytes / (1024 * 1024)
        );

        // --- allocate from it and export a dma-buf ----------------------------
        let allocate: libloading::Symbol<
            unsafe extern "C" fn(u64, usize, u32, *mut *mut c_void) -> Status,
        > = hsa.symbol(b"hsa_amd_memory_pool_allocate\0")?;
        let mut pointer: *mut c_void = std::ptr::null_mut();
        let status = allocate(pools.found.pool, bytes, 0, &raw mut pointer);
        if status != HSA_STATUS_SUCCESS {
            return Err(format!("hsa_amd_memory_pool_allocate returned {status}"));
        }
        println!("pool allocation: {mib} MiB at {pointer:p}");

        // The route that would make a shared buffer a first-class hrx buffer: allocate it
        // through the GPU runtime, take its device pointer, and export that.
        match hrx::Stream::open() {
            Ok(mut stream) => match stream.allocate_shared(bytes) {
                Ok(buffer) => {
                    stream.fill(buffer.binding(), 0x11).ok();
                    stream.synchronize().ok();
                    match buffer.device_ptr() {
                        Ok(hrx_pointer) => {
                            println!("hrx shared allocation: ok, device ptr {hrx_pointer:p}");
                            let mut fd: i32 = -1;
                            let mut offset: u64 = 0;
                            let export: libloading::Symbol<
                                unsafe extern "C" fn(
                                    *const c_void,
                                    usize,
                                    *mut i32,
                                    *mut u64,
                                ) -> Status,
                            > = hsa.symbol(b"hsa_amd_portable_export_dmabuf\0")?;
                            let status = export(hrx_pointer, bytes, &raw mut fd, &raw mut offset);
                            if status != HSA_STATUS_SUCCESS {
                                println!("  dma-buf export of an hrx buffer: NO (status {status})");
                            } else {
                                println!(
                                    "  dma-buf export of an hrx buffer: YES (fd {fd}, offset {offset})"
                                );
                                match dvxrt::Context::new(0, &xclbin) {
                                    Ok(context) => match context.import_dmabuf(fd, bytes) {
                                        Ok(bo) => {
                                            let mut seen = vec![0u8; 4096];
                                            bo.read(&mut seen).ok();
                                            let wrong = seen.iter().filter(|&&b| b != 0x11).count();
                                            println!(
                                                "  npu reads the hrx buffer: {}",
                                                if wrong == 0 {
                                                    "YES -- one buffer, both engines".into()
                                                } else {
                                                    format!("NO ({wrong}/4096 differ)")
                                                }
                                            );
                                        }
                                        Err(error) => println!("  npu import: NO ({error})"),
                                    },
                                    Err(error) => println!("  npu context: {error}"),
                                }
                            }
                        }
                        Err(error) => println!("hrx shared allocation device ptr: NO ({error})"),
                    }
                }
                Err(error) => println!("hrx shared allocation: NO ({error})"),
            },
            Err(error) => println!("hrx stream unavailable: {error}"),
        }

        let export: libloading::Symbol<
            unsafe extern "C" fn(*const c_void, usize, *mut i32, *mut u64) -> Status,
        > = hsa.symbol(b"hsa_amd_portable_export_dmabuf\0")?;
        let mut fd: i32 = -1;
        let mut offset: u64 = 0;
        let status = export(pointer, bytes, &raw mut fd, &raw mut offset);
        if status != HSA_STATUS_SUCCESS {
            println!("hsa dma-buf export of pool memory: NO (status {status})");
            return Ok(());
        }
        println!("hsa dma-buf export of pool memory: YES (fd {fd}, offset {offset})");

        // --- and hand that descriptor to the NPU ------------------------------
        match dvxrt::Context::new(0, &xclbin) {
            Ok(context) => {
                let _group = context.group_id(ARG_A).map_err(|e| e.to_string())?;
                match context.import_dmabuf(fd, bytes) {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        // Import alone only proves the descriptor was accepted. Write with
                        // the GPU and read through the NPU's BO to prove it is one buffer.
                        let fill: libloading::Symbol<
                            unsafe extern "C" fn(*mut c_void, u32, usize) -> Status,
                        > = hsa.symbol(b"hsa_amd_memory_fill\0")?;
                        let pattern = 0xa5a5_a5a5u32;
                        let status = fill(pointer, pattern, bytes / 4);
                        if status != HSA_STATUS_SUCCESS {
                            println!("  gpu fill of pool memory failed (status {status})");
                            return Ok(());
                        }
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != 0xa5).count();
                                if wrong == 0 {
                                    println!(
                                        "  gpu write -> npu read over dma-buf: YES -- zero-copy confirmed"
                                    );
                                } else {
                                    println!(
                                        "  gpu write -> npu read over dma-buf: NO ({wrong}/4096 bytes differ)"
                                    );
                                }
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("npu context: UNAVAILABLE ({error})"),
        }
    }
    Ok(())
}

pub fn copy(&self, dst: View<'_>, src: View<'_>) -> Result<()>

Queue a copy between equal, nonempty views owned by this stream.

Examples found in repository

examples/import_probe.rs (line 61)

fn main() -> Result<()> {
    let mib: usize = match std::env::args().collect::<Vec<_>>().as_slice() {
        [_] => 16,
        [_, flag, value] if flag == "--mib" => value.parse().expect("--mib expects an integer"),
        _ => {
            eprintln!("usage: import_probe [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");

    // Owned for the whole scope below; the imported buffer never frees it.
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");
    let result = (|| -> Result<()> {
        let mut stream = Stream::open()?;
        let buffer = match unsafe { stream.import_host(host.cast(), bytes) } {
            Ok(buffer) => buffer,
            Err(error) => {
                println!("import: UNSUPPORTED ({error})");
                return Ok(());
            }
        };
        println!("import: ok ({mib} MiB at {host:p})");

        // Prove the host pages start clear, so a match afterwards can only be the GPU.
        let before = unsafe { std::slice::from_raw_parts(host, bytes) };
        assert!(before.iter().all(|&b| b == 0), "host memory was not zeroed");

        stream.fill(buffer.binding(), PATTERN)?;
        stream.synchronize()?;

        // Read the ORIGINAL host pointer -- no download, no staging buffer.
        let after = unsafe { std::slice::from_raw_parts(host, bytes) };
        let wrong = after.iter().filter(|&&b| b != PATTERN).count();
        if wrong == 0 {
            println!("gpu fill visible on host pointer: YES (zero-copy confirmed)");
        } else {
            println!("gpu fill visible on host pointer: NO ({wrong}/{bytes} bytes unwritten)");
        }

        // A device-side read of the same buffer must also see host writes.
        let mirror = stream.allocate(bytes)?;
        unsafe { std::slice::from_raw_parts_mut(host, bytes) }.fill(0x3c);
        stream.copy(mirror.binding(), buffer.binding())?;
        let mut check = [0u8; 64];
        stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
        let coherent = check.iter().all(|&b| b == 0x3c);
        println!(
            "host write visible to device: {}",
            if coherent { "YES" } else { "NO" }
        );
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

examples/gpu_load.rs (line 108)

fn main() -> Result<()> {
    let options = match parse() {
        Ok(options) => options,
        Err(message) => {
            eprintln!("gpu_load: {message}");
            std::process::exit(2);
        }
    };
    let size = options.mib * 1024 * 1024;
    let mut stream = Stream::open()?;
    let source = stream.allocate(size)?;
    let destination = stream.allocate(size)?;
    // Touch both allocations so the steady-state loop never pays first-use faults.
    stream.fill(source.binding(), 0x3c)?;
    stream.fill(destination.binding(), 0)?;
    stream.synchronize()?;

    let run_batch = |stream: &mut Stream| -> Result<()> {
        for _ in 0..options.batch {
            match options.mode {
                Mode::Copy => stream.copy(destination.binding(), source.binding())?,
                Mode::Fill => stream.fill(destination.binding(), 0x5a)?,
            }
        }
        stream.synchronize()
    };

    // Warm up outside the measured window: the first batch pays queue and paging costs.
    run_batch(&mut stream)?;

    let window = Duration::from_secs_f64(options.seconds);
    let start = Instant::now();
    let mut iterations = 0usize;
    // Per-batch rates, so the harness can see whether contention was steady or bursty.
    let mut batch_rates = Vec::new();
    while start.elapsed() < window {
        let batch_start = Instant::now();
        run_batch(&mut stream)?;
        let seconds = batch_start.elapsed().as_secs_f64();
        iterations += options.batch;
        let bytes = options.mode.traffic(size) * options.batch;
        batch_rates.push(bytes as f64 / seconds / 1e9);
    }
    let elapsed = start.elapsed().as_secs_f64();
    let total = options.mode.traffic(size) as f64 * iterations as f64;
    let mean = total / elapsed / 1e9;
    batch_rates.sort_by(f64::total_cmp);
    let median = batch_rates[batch_rates.len() / 2];
    let low = batch_rates[batch_rates.len() / 20];

    println!(
        concat!(
            r#"{{"device":"gpu","mode":"{}","buffer_mib":{},"batch":{},"#,
            r#""seconds":{:.4},"iterations":{},"gb_s_mean":{:.3},"#,
            r#""gb_s_median":{:.3},"gb_s_p5":{:.3}}}"#
        ),
        options.mode.name(),
        options.mib,
        options.batch,
        elapsed,
        iterations,
        mean,
        median,
        low
    );
    eprintln!(
        "gpu_load: {} {:.1} GB/s mean over {:.1}s ({} iterations)",
        options.mode.name(),
        mean,
        elapsed,
        iterations
    );
    Ok(())
}

examples/unified_probe.rs (line 200)

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().expect("--mib expects an integer"),
        ),
        _ => {
            eprintln!("usage: unified_probe <xclbin> [--mib N]");
            std::process::exit(2);
        }
    };
    let bytes = mib * 1024 * 1024;
    let layout = Layout::from_size_align(bytes, PAGE).expect("page-aligned layout");
    let host = unsafe { alloc_zeroed(layout) };
    assert!(!host.is_null(), "host allocation failed");

    let result = (|| -> Result<()> {
        println!("host allocation: {mib} MiB at {host:p}");

        let context = match unsafe { dvxrt::Context::new(0, &xclbin) } {
            Ok(context) => context,
            Err(error) => {
                println!("npu context: UNAVAILABLE ({error})");
                return Ok(());
            }
        };
        let group = context.group_id(ARG_A).expect("group id for A");

        // 1. Can amdxdna pin pages we allocated? Reported either way -- if it can, either
        //    runtime may own the allocation; if it cannot, only the NPU's allocator may.
        match unsafe { context.import_bo(host.cast(), bytes, dvxrt::BoKind::HostOnly, group) } {
            Ok(_) => println!("npu userptr import (host-owned pages): YES"),
            Err(error) => println!("npu userptr import (host-owned pages): NO ({error})"),
        }

        // 2. The direction that matters: let the NPU's allocator own the pages -- it is the
        //    more constrained of the two -- and import its host mapping into the GPU runtime.
        let bo = context
            .alloc_bo(bytes, dvxrt::BoKind::HostOnly, group)
            .expect("allocate an NPU BO");
        let shared = bo.map().expect("map the NPU BO");
        println!("npu allocation mapped at {shared:p}");
        assert_eq!(shared as usize % PAGE, 0, "NPU mapping is not page-aligned");

        let mut stream = Stream::open()?;
        let imported = unsafe { stream.import_host(shared.cast(), bytes) };
        match &imported {
            Ok(_) => println!("gpu import of the NPU mapping: YES"),
            Err(error) => println!("gpu import of the NPU mapping: NO ({error})"),
        }

        // 3. If the shared mapping took, check that a GPU write lands in the NPU's BO.
        if let Ok(buffer) = &imported {
            stream.fill(buffer.binding(), GPU_PATTERN)?;
            stream.synchronize()?;
            bo.sync(false, bytes).expect("invalidate for device reads");
            let mut seen = vec![0u8; 4096];
            bo.read(&mut seen).expect("read through the NPU BO");
            let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
            println!(
                "gpu write -> npu read: {}",
                if wrong == 0 {
                    "YES".into()
                } else {
                    format!("NO ({wrong} bytes differ)")
                }
            );
        }

        // 3b. The dma-buf route: the GPU allocates, HSA exports the allocation as a
        //     dma-buf, and XRT imports that descriptor. Neither runtime copies, and unlike
        //     the host-pointer routes above neither driver has to pin the other's pages.
        // The GPU-side import above HSA-locked these pages, so they are exportable.
        let locked = unsafe { stream.import_host(host.cast(), bytes) };
        let export = match &locked {
            Ok(buffer) => {
                stream.fill(buffer.binding(), GPU_PATTERN)?;
                stream.synchronize()?;
                export_dmabuf(host.cast(), bytes)
            }
            Err(error) => Err(format!("host pages are not GPU-resident: {error}")),
        };
        match export {
            Ok((fd, size)) => {
                println!("hsa dma-buf export of a GPU buffer: YES (fd {fd})");
                match unsafe { context.import_dmabuf(fd, size) } {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != GPU_PATTERN).count();
                                println!(
                                    "  gpu write -> npu read over dma-buf: {}",
                                    if wrong == 0 {
                                        "YES".into()
                                    } else {
                                        format!("NO ({wrong} bytes differ)")
                                    }
                                );
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("hsa dma-buf export of a GPU buffer: NO ({error})"),
        }

        // 4. And the reverse: the NPU's BO writes, the GPU reads.
        if let Ok(buffer) = &imported {
            bo.write(&vec![NPU_PATTERN; bytes])
                .expect("write through the NPU BO");
            let mirror = stream.allocate(bytes)?;
            stream.copy(mirror.binding(), buffer.binding())?;
            let mut check = [0u8; 64];
            stream.read_blocking(mirror.binding().slice(0, check.len())?, &mut check)?;
            let back = check.iter().all(|&b| b == NPU_PATTERN);
            println!("npu write -> gpu read: {}", if back { "YES" } else { "NO" });
        }
        Ok(())
    })();
    unsafe { dealloc(host, layout) };
    result
}

pub fn upload(&mut self, dst: View<'_>, bytes: &[u8]) -> Result<()>

Copy the input into runtime-owned staging, then enqueue an actual GPU buffer copy. The caller’s slice is no longer referenced when this returns. Bytes are written at the start of the view and must fit within it. Staging pressure may submit or wait for prior work to bound memory use.

Examples found in repository

examples/stream_bench.rs (line 60)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn upload_at( &mut self, dst: &Buffer, offset: usize, bytes: &[u8], ) -> Result<()>

Enqueue an upload at a checked byte offset into an allocation.

pub fn allocate_shared(&self, bytes: usize) -> Result<Buffer>

Allocate a host-local, device-visible buffer.

Unlike Stream::allocate, the runtime hands out a device pointer for these, so the allocation can be exported to another driver – which is what lets one buffer serve both the GPU and the NPU. Device-local memory is faster for GPU-only work.

Examples found in repository

examples/dmabuf_probe.rs (line 192)

fn main() -> Result<(), String> {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let (xclbin, mib) = match args.as_slice() {
        [xclbin] => (xclbin.clone(), 4usize),
        [xclbin, flag, value] if flag == "--mib" => (
            xclbin.clone(),
            value.parse().map_err(|_| "--mib expects an integer")?,
        ),
        _ => return Err("usage: dmabuf_probe <xclbin> [--mib N]".into()),
    };
    let bytes = mib * 1024 * 1024;
    let hsa = Hsa::open()?;

    unsafe {
        let init: libloading::Symbol<unsafe extern "C" fn() -> Status> =
            hsa.symbol(b"hsa_init\0")?;
        let status = init();
        if status != HSA_STATUS_SUCCESS {
            return Err(format!("hsa_init returned {status}"));
        }

        // --- find a GPU agent -------------------------------------------------
        let iterate_agents: libloading::Symbol<
            unsafe extern "C" fn(
                unsafe extern "C" fn(u64, *mut c_void) -> Status,
                *mut c_void,
            ) -> Status,
        > = hsa.symbol(b"hsa_iterate_agents\0")?;
        let agent_info: libloading::Symbol<unsafe extern "C" fn(u64, u32, *mut c_void) -> Status> =
            hsa.symbol(b"hsa_agent_get_info\0")?;

        // The callback needs the symbol; pass it through the user-data pointer.
        struct AgentScan<'a> {
            info: &'a dyn Fn(u64, u32, *mut c_void) -> Status,
            found: Found,
        }
        unsafe extern "C" fn on_agent(agent: u64, data: *mut c_void) -> Status {
            let scan = unsafe { &mut *data.cast::<AgentScan>() };
            let mut kind: u32 = 0;
            if (scan.info)(agent, HSA_AGENT_INFO_DEVICE, (&raw mut kind).cast())
                == HSA_STATUS_SUCCESS
                && kind == HSA_DEVICE_TYPE_GPU
                && scan.found.agent == 0
            {
                scan.found.agent = agent;
            }
            HSA_STATUS_SUCCESS
        }
        let info_fn = |agent, key, out| agent_info(agent, key, out);
        let mut scan = AgentScan {
            info: &info_fn,
            found: Found::default(),
        };
        iterate_agents(on_agent, (&raw mut scan).cast());
        if scan.found.agent == 0 {
            return Err("no GPU agent found".into());
        }
        println!("gpu agent: 0x{:x}", scan.found.agent);

        // --- find an allocatable global pool ----------------------------------
        let iterate_pools: libloading::Symbol<
            unsafe extern "C" fn(
                u64,
                unsafe extern "C" fn(u64, *mut c_void) -> Status,
                *mut c_void,
            ) -> Status,
        > = hsa.symbol(b"hsa_amd_agent_iterate_memory_pools\0")?;
        let pool_info: libloading::Symbol<unsafe extern "C" fn(u64, u32, *mut c_void) -> Status> =
            hsa.symbol(b"hsa_amd_memory_pool_get_info\0")?;

        struct PoolScan<'a> {
            info: &'a dyn Fn(u64, u32, *mut c_void) -> Status,
            found: Found,
        }
        unsafe extern "C" fn on_pool(pool: u64, data: *mut c_void) -> Status {
            let scan = unsafe { &mut *data.cast::<PoolScan>() };
            if scan.found.pool != 0 {
                return HSA_STATUS_SUCCESS;
            }
            let mut segment: u32 = u32::MAX;
            let mut allowed: bool = false;
            let mut size: usize = 0;
            (scan.info)(pool, POOL_INFO_SEGMENT, (&raw mut segment).cast());
            (scan.info)(pool, POOL_INFO_ALLOC_ALLOWED, (&raw mut allowed).cast());
            (scan.info)(pool, POOL_INFO_SIZE, (&raw mut size).cast());
            if segment == SEGMENT_GLOBAL && allowed && size > 0 {
                scan.found.pool = pool;
                scan.found.pool_bytes = size;
            }
            HSA_STATUS_SUCCESS
        }
        let pool_fn = |pool, key, out| pool_info(pool, key, out);
        let mut pools = PoolScan {
            info: &pool_fn,
            found: Found::default(),
        };
        iterate_pools(scan.found.agent, on_pool, (&raw mut pools).cast());
        if pools.found.pool == 0 {
            return Err("no allocatable global memory pool on the GPU agent".into());
        }
        println!(
            "gpu memory pool: 0x{:x} ({} MiB)",
            pools.found.pool,
            pools.found.pool_bytes / (1024 * 1024)
        );

        // --- allocate from it and export a dma-buf ----------------------------
        let allocate: libloading::Symbol<
            unsafe extern "C" fn(u64, usize, u32, *mut *mut c_void) -> Status,
        > = hsa.symbol(b"hsa_amd_memory_pool_allocate\0")?;
        let mut pointer: *mut c_void = std::ptr::null_mut();
        let status = allocate(pools.found.pool, bytes, 0, &raw mut pointer);
        if status != HSA_STATUS_SUCCESS {
            return Err(format!("hsa_amd_memory_pool_allocate returned {status}"));
        }
        println!("pool allocation: {mib} MiB at {pointer:p}");

        // The route that would make a shared buffer a first-class hrx buffer: allocate it
        // through the GPU runtime, take its device pointer, and export that.
        match hrx::Stream::open() {
            Ok(mut stream) => match stream.allocate_shared(bytes) {
                Ok(buffer) => {
                    stream.fill(buffer.binding(), 0x11).ok();
                    stream.synchronize().ok();
                    match buffer.device_ptr() {
                        Ok(hrx_pointer) => {
                            println!("hrx shared allocation: ok, device ptr {hrx_pointer:p}");
                            let mut fd: i32 = -1;
                            let mut offset: u64 = 0;
                            let export: libloading::Symbol<
                                unsafe extern "C" fn(
                                    *const c_void,
                                    usize,
                                    *mut i32,
                                    *mut u64,
                                ) -> Status,
                            > = hsa.symbol(b"hsa_amd_portable_export_dmabuf\0")?;
                            let status = export(hrx_pointer, bytes, &raw mut fd, &raw mut offset);
                            if status != HSA_STATUS_SUCCESS {
                                println!("  dma-buf export of an hrx buffer: NO (status {status})");
                            } else {
                                println!(
                                    "  dma-buf export of an hrx buffer: YES (fd {fd}, offset {offset})"
                                );
                                match dvxrt::Context::new(0, &xclbin) {
                                    Ok(context) => match context.import_dmabuf(fd, bytes) {
                                        Ok(bo) => {
                                            let mut seen = vec![0u8; 4096];
                                            bo.read(&mut seen).ok();
                                            let wrong = seen.iter().filter(|&&b| b != 0x11).count();
                                            println!(
                                                "  npu reads the hrx buffer: {}",
                                                if wrong == 0 {
                                                    "YES -- one buffer, both engines".into()
                                                } else {
                                                    format!("NO ({wrong}/4096 differ)")
                                                }
                                            );
                                        }
                                        Err(error) => println!("  npu import: NO ({error})"),
                                    },
                                    Err(error) => println!("  npu context: {error}"),
                                }
                            }
                        }
                        Err(error) => println!("hrx shared allocation device ptr: NO ({error})"),
                    }
                }
                Err(error) => println!("hrx shared allocation: NO ({error})"),
            },
            Err(error) => println!("hrx stream unavailable: {error}"),
        }

        let export: libloading::Symbol<
            unsafe extern "C" fn(*const c_void, usize, *mut i32, *mut u64) -> Status,
        > = hsa.symbol(b"hsa_amd_portable_export_dmabuf\0")?;
        let mut fd: i32 = -1;
        let mut offset: u64 = 0;
        let status = export(pointer, bytes, &raw mut fd, &raw mut offset);
        if status != HSA_STATUS_SUCCESS {
            println!("hsa dma-buf export of pool memory: NO (status {status})");
            return Ok(());
        }
        println!("hsa dma-buf export of pool memory: YES (fd {fd}, offset {offset})");

        // --- and hand that descriptor to the NPU ------------------------------
        match dvxrt::Context::new(0, &xclbin) {
            Ok(context) => {
                let _group = context.group_id(ARG_A).map_err(|e| e.to_string())?;
                match context.import_dmabuf(fd, bytes) {
                    Ok(imported) => {
                        println!("npu import of the GPU dma-buf: YES");
                        // Import alone only proves the descriptor was accepted. Write with
                        // the GPU and read through the NPU's BO to prove it is one buffer.
                        let fill: libloading::Symbol<
                            unsafe extern "C" fn(*mut c_void, u32, usize) -> Status,
                        > = hsa.symbol(b"hsa_amd_memory_fill\0")?;
                        let pattern = 0xa5a5_a5a5u32;
                        let status = fill(pointer, pattern, bytes / 4);
                        if status != HSA_STATUS_SUCCESS {
                            println!("  gpu fill of pool memory failed (status {status})");
                            return Ok(());
                        }
                        let mut seen = vec![0u8; 4096];
                        match imported.read(&mut seen) {
                            Ok(()) => {
                                let wrong = seen.iter().filter(|&&b| b != 0xa5).count();
                                if wrong == 0 {
                                    println!(
                                        "  gpu write -> npu read over dma-buf: YES -- zero-copy confirmed"
                                    );
                                } else {
                                    println!(
                                        "  gpu write -> npu read over dma-buf: NO ({wrong}/4096 bytes differ)"
                                    );
                                }
                            }
                            Err(error) => {
                                println!("  read through the imported BO failed: {error}")
                            }
                        }
                    }
                    Err(error) => println!("npu import of the GPU dma-buf: NO ({error})"),
                }
            }
            Err(error) => println!("npu context: UNAVAILABLE ({error})"),
        }
    }
    Ok(())
}

pub fn submit(&mut self) -> Result<Submission<'_>>

Submit pending commands before querying completion; native query alone does not include the unsubmitted command buffer.

pub unsafe fn load(&self, path: &Path, symbol: &str) -> Result<Kernel>

Load a native code object and select one export by name.

Safety

The code object must be trusted native machine code.

pub unsafe fn load_artifact(&self, artifact: &Artifact) -> Result<Kernel>

Load a compiled artifact directly, without a filesystem round trip.

Safety

The artifact must be trusted native code, as for Stream::load.

Examples found in repository

examples/gemm_pipeline.rs (line 37)

fn gpu_kernel(
    runtime: &Runtime,
    source: &str,
    symbol: &str,
    config: &[(&str, String)],
    elements: usize,
    bindings: Vec<BindingContract>,
) -> Result<GpuKernel> {
    let compiler = hrx::loom::Compiler::for_target(None, runtime.gpu()?.target())?;
    let module = compiler.module(source);
    let mut spec = hrx::loom::Specialization::new(symbol);
    for (key, value) in config {
        spec.set_config(*key, value.clone());
    }
    let artifact = module.compile(&spec)?;
    let stream = hrx::gpu::Stream::open()?;
    let raw = unsafe { stream.load_artifact(&artifact) }?;
    let mut constants = hrx::gpu::Constants::new();
    match raw.info().constant_byte_length {
        8 => constants.push(elements as u32)?,
        12 => constants.push(elements as u64)?,
        _ => return Err(hrx::Error::Message("unexpected scalar ABI".into())),
    }
    constants.push(1f32)?;
    unsafe {
        runtime.load_gpu_kernel(
            artifact.path(),
            symbol,
            [elements.div_ceil(256) as u32, 1, 1],
            [256, 1, 1],
            KernelContract {
                bindings,
                constants: constants.as_bytes().to_vec(),
            },
        )
    }
}

examples/stream_bench.rs (line 83)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub unsafe fn dispatch( &self, kernel: &Kernel, grid: [u32; 3], block: [u32; 3], constants: &Constants, bindings: &[View<'_>], ) -> Result<()>

Queue a kernel invocation with explicitly packed constants and borrowed bindings.

Safety

Kernel, dimensions, constants and binding spans must agree. GPU addressing is not sandboxed by a binding’s length.

Examples found in repository

examples/stream_bench.rs (line 101)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn scratch(&mut self, bytes: usize) -> Result<Buffer>

Reuse a cached allocation or allocate new scratch storage. Reuse follows the stream’s command order and requires no host wait. Return it explicitly with Stream::recycle after recording its last use. Dropping a buffer releases it rather than returning it to this pool.

Examples found in repository

examples/stream_bench.rs (line 40)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn recycle(&mut self, buffer: Buffer) -> Result<()>

Return an allocation to this stream’s bounded scratch pool after recording its final use. A later scratch request may reuse its storage. Returns an error for a buffer another stream allocated: a scratch pool is one stream’s private free list, even though the buffer itself is usable from any stream on the device. Recycling needs mutable stream access; there is no automatic return on drop.

Examples found in repository

examples/stream_bench.rs (line 42)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn set_scratch_limit(&mut self, bytes: usize)

Set the scratch byte budget, evicting the largest size classes as needed.

impl Stream

pub fn graph(&self) -> Result<Graph<'_>>

Begin recording while borrowing this stream and the recorded resources. finish instantiates an owned executable and ends those borrows.

fn escape() -> hrx::Result<hrx::Graph<'static>> {
    let stream = hrx::Stream::open()?;
    stream.graph()
}

Examples found in repository

examples/graph_prepare.rs (line 18)

fn main() -> hrx::Result<()> {
    let runtime = Runtime::new()?;
    let buffer = runtime.allocate(4096, MemoryPlacement::GpuLocal)?;
    let stream = Stream::open()?;
    let raw = stream.allocate(4096)?;
    for count in [128, 512, 1024, 2048] {
        let begin = Instant::now();
        let mut graph = runtime.graph();
        for _ in 0..count {
            graph.fill(buffer.view(), 0)?;
        }
        let graph = graph.prepare()?;
        let inferred = begin.elapsed();
        let begin = Instant::now();
        let mut native = stream.graph()?;
        let mut last = None;
        for _ in 0..count {
            last = Some(native.fill(last.as_slice(), raw.binding(), 0)?);
        }
        let _native = native.finish()?;
        println!(
            "nodes={count} inferred_us={} explicit_us={}",
            inferred.as_micros(),
            begin.elapsed().as_micros()
        );
        graph.submit()?.wait()?;
    }
    Ok(())
}

examples/stream_bench.rs (line 120)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

pub fn launch(&mut self, graph: &mut GraphExec) -> Result<()>

Replay an instantiated graph on its original stream.

A native submission failure may leave partial work in flight. Such a graph cannot be replayed again and its native resources are retained on drop because completion is unknown. Rejecting a foreign stream does not invalidate the graph.

Examples found in repository

examples/stream_bench.rs (line 160)

fn main() -> Result<()> {
    let mut stream = Stream::open()?;
    let mut metrics = BTreeMap::new();
    let samples = std::env::var("HRX_BENCH_SAMPLES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .filter(|&value| value > 0)
        .unwrap_or(9);
    let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
        for _ in 0..50_000 {
            let buffer = stream.scratch(1024 * 1024)?;
            black_box(&buffer);
            stream.recycle(buffer)?;
        }
        Ok(())
    })?;
    metrics.insert("scratch_reuse_ns", ns);
    let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
        for _ in 0..2_000 {
            black_box(stream.allocate(1024 * 1024)?);
        }
        Ok(())
    })?;
    metrics.insert("allocate_drop_ns", ns);

    let bytes = 256 * 1024 * 1024;
    let chunk = vec![0x3cu8; 4 * 1024 * 1024];
    let weights = stream.allocate(bytes)?;
    let (_, ns) = measure(&mut stream, samples, 1, |stream| {
        for offset in (0..bytes).step_by(chunk.len()) {
            stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
        }
        Ok(())
    })?;
    metrics.insert(
        "queued_upload_gib_s",
        bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
    );
    let mut check = [0u8; 16];
    stream.read_blocking(
        weights.try_slice(bytes - check.len(), check.len())?,
        &mut check,
    )?;
    assert_eq!(check, [0x3c; 16]);
    drop(weights);

    let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
    let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
    let mut spec = hrx::loom::Specialization::new("krea2_euler");
    spec.set_config("krea2.euler.grid_x", "1");
    spec.set_config("krea2.euler.grid_y", "1");
    let artifact = module.compile(&spec)?;
    // This repository owns the source; use its exact dimensions and scalar layout.
    let kernel = unsafe { stream.load_artifact(&artifact)? };
    let sample = stream.allocate(512)?;
    let velocity = stream.allocate(512)?;
    let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
    stream.upload_blocking(sample.binding(), &ones)?;
    stream.upload_blocking(velocity.binding(), &ones)?;
    let mut constants = Constants::new();
    match kernel.info().constant_byte_length {
        8 => constants.push(256u32)?,
        12 => constants.push(256u64)?,
        n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
    }
    constants.push(0f32)?;
    let bindings = [sample.binding(), velocity.binding()];
    let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
        for _ in 0..2048 {
            // The source accesses exactly the 256 bf16 elements in each buffer.
            unsafe {
                stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
            }
        }
        Ok(())
    })?;
    metrics.insert("dispatch_enqueue_ns", enqueue);
    metrics.insert("dispatch_complete_ns", complete);

    // Use disjoint allocations in both graphs. Even an idempotent store is a
    // write: concurrent Euler dispatches cannot share their in/out buffer.
    let mut graph_buffers = Vec::new();
    for _ in 0..32 {
        let sample = stream.allocate(512)?;
        let velocity = stream.allocate(512)?;
        stream.upload(sample.binding(), &ones)?;
        stream.upload(velocity.binding(), &ones)?;
        graph_buffers.push((sample, velocity));
    }
    stream.synchronize()?;
    let mut builder = stream.graph()?;
    let mut previous = None;
    for (sample, velocity) in &graph_buffers {
        let after = previous.as_slice();
        previous = Some(unsafe {
            builder.dispatch(
                after,
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?
        });
    }
    let chained = builder.finish()?;

    // The same disjoint workloads, this time allowing concurrent scheduling.
    let mut builder = stream.graph()?;
    for (sample, velocity) in &graph_buffers {
        unsafe {
            builder.dispatch(
                &[],
                &kernel,
                [1; 3],
                [256, 1, 1],
                &constants,
                &[sample.binding(), velocity.binding()],
            )?;
        }
    }
    let independent = builder.finish()?;
    let mut graphs = [chained, independent];
    let mut enqueue = [Vec::new(), Vec::new()];
    let mut complete = [Vec::new(), Vec::new()];
    for round in 0..samples + 3 {
        for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
            stream.synchronize()?;
            let start = Instant::now();
            for _ in 0..256 {
                stream.launch(&mut graphs[arm])?;
            }
            let submitted = start.elapsed();
            stream.synchronize()?;
            let elapsed = start.elapsed();
            if round >= 3 {
                enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
                complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
            }
        }
    }
    for values in enqueue.iter_mut().chain(complete.iter_mut()) {
        values.sort_by(f64::total_cmp);
    }
    let ns = complete[0][samples / 2];
    let ns_free = complete[1][samples / 2];
    metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
    metrics.insert("graph_complete_ns_per_kernel", ns);
    metrics.insert("graph_independent_ns_per_kernel", ns_free);
    metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
    for (sample, _) in &graph_buffers {
        let mut output = vec![0; 512];
        stream.read_blocking(sample.binding(), &mut output)?;
        assert_eq!(output, ones);
    }
    let mut output = vec![0; 512];
    stream.read_blocking(sample.binding(), &mut output)?;
    assert_eq!(output, ones);
    println!("{}", serde_json::to_string(&metrics)?);
    Ok(())
}

impl Stream

pub fn read(&mut self, source: View<'_>) -> Result<Readback>

Queue a download into owned, initially unmapped host-visible storage.

Trait Implementations

impl Debug for Stream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for Stream

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.