Struct Completion 

pub struct Completion { /* private fields */ }

An owned completion observer. Dropping it does not cancel submitted work.

Polling never waits for a device. Clones may be waited on by different tasks.

Implementations

impl Completion

pub fn finished_at(&self) -> Option<Instant>

Host monotonic time at which execution reached a terminal state. This permits pipeline boundary timing without including a caller’s late wait. It is not a device timestamp and does not imply successful execution.

pub fn profile(&self) -> Option<CompletionProfile>

Inspect final host-observed timing without waiting for completion.

pub fn result(&self) -> Option<Result<()>>

Observe a terminal result without blocking, or None while pending.

pub fn is_complete(&self) -> bool

Whether the operation reached a terminal state, including failure.

pub fn wait(&self) -> Result<()>

Wait for device completion, propagating execution failure. The caller may execute ready regions from this submission to avoid a worker handoff. Engine limits and all dependency checks still apply.

Examples found in repository

examples/graph_prepare.rs (line 29)

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/shared_roundtrip.rs (line 46)

fn main() -> Result<()> {
    let args: Vec<_> = std::env::args().collect();
    if args.len() != 4 {
        return Err(hrx::Error::Message(
            "usage: shared_roundtrip <passthrough.xclbin> <instructions.bin> <bytes>".into(),
        ));
    }
    let bytes = args[3]
        .parse::<usize>()
        .map_err(|e| hrx::Error::Message(e.to_string()))?;
    let runtime = Runtime::new()?;
    // This example accepts trusted local artifacts with the documented ABI.
    let program = unsafe { runtime.npu(0)?.load_program(&args[1]) }?;
    let binding = |bytes, access| BindingContract {
        bytes,
        access,
        alignment: 4,
        layout: "u32 passthrough".into(),
    };
    let contract = KernelContract {
        bindings: vec![
            binding(bytes, Access::Read),
            binding(4096, Access::Read),
            binding(bytes, Access::Write),
        ],
        constants: vec![],
    };
    let kernel = unsafe { program.kernel(&std::fs::read(&args[2])?, contract) }?;
    let a = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
    let b = runtime.allocate(4096, MemoryPlacement::Shared(program.clone()))?;
    let c = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
    let result = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
    let mut graph = runtime.graph();
    graph.fill(a.view(), 0x5a)?;
    graph.npu(&kernel, &[a.view(), b.view(), c.view()])?;
    graph.copy(result.view(), c.view())?;
    let graph = graph.prepare()?;
    for _ in 0..10 {
        graph.submit()?.wait()?;
        let mapped = result.map_read()?;
        if mapped.iter().any(|&b| b != 0x5a) {
            return Err(hrx::Error::Message("GPU/NPU roundtrip mismatch".into()));
        }
    }
    println!(
        "GPU -> NPU -> GPU: all {bytes} bytes correct, 10 executions, {} prepared regions",
        graph.regions()
    );
    Ok(())
}

examples/shared_bench.rs (line 48)

fn main() -> Result<()> {
    let args: Vec<_> = std::env::args().collect();
    if args.len() != 4 {
        return Err(hrx::Error::Message(
            "usage: shared_bench <passthrough.xclbin> <instructions.bin> <bytes>".into(),
        ));
    }
    let bytes = args[3]
        .parse::<usize>()
        .map_err(|e| hrx::Error::Message(e.to_string()))?;
    let runtime = Runtime::new()?;
    let program = unsafe { runtime.npu(0)?.load_program(&args[1]) }?;
    let binding = |bytes, access| BindingContract {
        bytes,
        access,
        alignment: 4,
        layout: "passthrough".into(),
    };
    let kernel = unsafe {
        program.kernel(
            &std::fs::read(&args[2])?,
            KernelContract {
                bindings: vec![
                    binding(bytes, Access::Read),
                    binding(4096, Access::Read),
                    binding(bytes, Access::Write),
                ],
                constants: vec![],
            },
        )
    }?;
    let a = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
    let b = runtime.allocate(4096, MemoryPlacement::Shared(program.clone()))?;
    let c = runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()))?;
    let mut graph = runtime.graph();
    graph.fill(a.view(), 0x35)?;
    graph.npu(&kernel, &[a.view(), b.view(), c.view()])?;
    let graph = graph.prepare()?;
    for _ in 0..5 {
        graph.submit()?.wait()?;
    }
    let mut samples = Vec::new();
    for _ in 0..100 {
        let start = Instant::now();
        graph.submit()?.wait()?;
        samples.push(start.elapsed().as_secs_f64());
    }
    if c.map_read()?.iter().any(|&b| b != 0x35) {
        return Err(hrx::Error::Message("incorrect benchmark output".into()));
    }
    samples.sort_by(f64::total_cmp);
    println!(
        "bytes={bytes} p50_us={:.2} p95_us={:.2}",
        percentile(&samples, 50) * 1e6,
        percentile(&samples, 95) * 1e6
    );
    Ok(())
}

examples/gemm_pipeline.rs (line 136)

fn main() -> Result<()> {
    let args: Vec<_> = std::env::args().collect();
    if args.len() != 6 {
        return Err(hrx::Error::Message(
            "usage: gemm_pipeline <row-major-bf16.xclbin> <instructions.bin> M K N".into(),
        ));
    }
    let parse = |i: usize| {
        args[i]
            .parse::<usize>()
            .map_err(|e| hrx::Error::Message(e.to_string()))
    };
    let (m, k, n) = (parse(3)?, parse(4)?, parse(5)?);
    let extent = |a: usize, b: usize, width: usize| {
        a.checked_mul(b)
            .filter(|&count| count > 0 && count <= (1 << 30))
            .and_then(|count| count.checked_mul(width))
            .ok_or_else(|| hrx::Error::Message("invalid matrix extent".into()))
    };
    let (a_bytes, b_bytes, c_bytes) = (extent(m, k, 2)?, extent(k, n, 2)?, extent(m, n, 4)?);
    let runtime = Runtime::new()?;
    let program = unsafe { runtime.npu(0)?.load_program(&args[1]) }?;
    let npu = unsafe {
        program.kernel(
            &std::fs::read(&args[2])?,
            KernelContract {
                bindings: vec![
                    binding(a_bytes, Access::Read),
                    binding(b_bytes, Access::Read),
                    binding(c_bytes, Access::Write),
                ],
                constants: vec![],
            },
        )
    }?;
    let prep = gpu_kernel(
        &runtime,
        include_str!("../tests/kernels/euler.loom"),
        "krea2_euler",
        &[
            ("krea2.euler.grid_x", (m * k).div_ceil(256).to_string()),
            ("krea2.euler.grid_y", "1".into()),
        ],
        m * k,
        vec![
            binding(a_bytes, Access::ReadWrite),
            binding(a_bytes, Access::Read),
        ],
    )?;
    let epilogue = gpu_kernel(
        &runtime,
        include_str!("../tests/kernels/add_f32.loom"),
        "add_f32",
        &[("add.grid", (m * n).div_ceil(256).to_string())],
        m * n,
        vec![binding(c_bytes, Access::ReadWrite)],
    )?;
    let allocate = |bytes| runtime.allocate(bytes, MemoryPlacement::Shared(program.clone()));
    let weights = allocate(b_bytes)?;
    let ones = allocate(a_bytes)?;
    for buffer in [&weights, &ones] {
        for word in buffer.map_write()?.as_chunks_mut::<2>().0 {
            word.copy_from_slice(&0x3f80u16.to_le_bytes());
        }
    }
    let prepare = || -> Result<(ExecutableGraph, Buffer)> {
        let a = allocate(a_bytes)?;
        let c = allocate(c_bytes)?;
        let mut graph = runtime.graph();
        graph.fill(a.view(), 0)?;
        graph.gpu(&prep, &[a.view(), ones.view()])?;
        graph.npu(&npu, &[a.view(), weights.view(), c.view()])?;
        graph.gpu(&epilogue, &[c.view()])?;
        Ok((graph.prepare()?, c))
    };
    let (first, c0) = prepare()?;
    let (second, c1) = prepare()?;
    for _ in 0..5 {
        first.submit()?.wait()?;
        second.submit()?.wait()?;
    }
    let before = runtime.statistics();
    let mut latency = Vec::with_capacity(50);
    for _ in 0..50 {
        let start = Instant::now();
        first.submit()?.wait()?;
        latency.push(start.elapsed().as_secs_f64());
    }
    let start = Instant::now();
    for _ in 0..50 {
        let a = first.submit()?;
        let b = second.submit()?;
        a.wait()?;
        b.wait()?;
    }
    let requests_per_second = 100.0 / start.elapsed().as_secs_f64();
    for c in [&c0, &c1] {
        if c.map_read()?
            .as_chunks::<4>()
            .0
            .iter()
            .any(|v| f32::from_le_bytes(*v) != k as f32 + 1.0)
        {
            return Err(hrx::Error::Message("GEMM pipeline output mismatch".into()));
        }
    }
    let after = runtime.statistics();
    assert_eq!(before.allocations, after.allocations);
    assert_eq!(before.imports, after.imports);
    assert_eq!(after.copied_bytes, 0);
    latency.sort_by(f64::total_cmp);
    println!(
        "M={m} K={k} N={n} p50_ms={:.3} p95_ms={:.3} pipelined_requests_s={requests_per_second:.2}",
        percentile(&latency, 50) * 1e3,
        percentile(&latency, 95) * 1e3
    );
    println!("{}", serde_json::to_string(&after)?);
    Ok(())
}

pub fn wait_timeout(&self, timeout: Duration) -> Result<bool>

Wait at most timeout. false leaves execution and ownership intact.

pub fn cancel(&self)

Cancel unscheduled nodes. Running native work is drained before completion. This request does not revoke resources or abort an unrelated submission.

Trait Implementations

impl Clone for Completion

fn clone(&self) -> Completion

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Future for Completion

type Output = Result<(), Error>

The type of value produced on completion.

fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output>

Attempts to resolve the future to a final value, registering the current task for wakeup if the value is not yet available.