Struct Graph 

pub struct Graph<'a> { /* private fields */ }

Owned native commands being recorded under checked dependency edges.

Implementations

impl<'a> Graph<'a>

pub fn binding_bytes(&self) -> usize

Total exposed bytes of distinct allocations referenced by recorded nodes. Repeated views and pooled reuse of the same backing count only once.

pub fn fill( &mut self, after: &[Node], dst: View<'_>, pattern: u8, ) -> Result<Node>

Record a byte fill after checked predecessor nodes.

Examples found in repository

examples/graph_prepare.rs (line 21)

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(())
}

pub fn copy( &mut self, after: &[Node], dst: View<'_>, src: View<'_>, ) -> Result<Node>

Record a non-overlapping byte copy after checked predecessor nodes.

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

Record a fixed native kernel invocation.

Safety

Kernel accesses must obey bindings and constants; dependency edges must order conflicting accesses for every replay.

Examples found in repository

examples/patch_bench.rs (lines 307-314)

fn prepare(
    stream: &mut Stream,
    compiler: &Compiler,
    case: &Case,
    shared: Option<std::sync::Arc<Vec<Buffer>>>,
) -> Result<Arm> {
    let artifact = compiler.module(case.source).compile(&case.spec)?;
    let kernel = unsafe { stream.load_artifact(&artifact) }?;
    let constants = Constants::indices(&kernel, &case.indices)?;
    let buffers = if let Some(buffers) = shared {
        buffers
    } else {
        std::sync::Arc::new(
            case.data
                .iter()
                .map(|data| {
                    let buffer = stream.allocate(data.len())?;
                    stream.upload_blocking(buffer.binding(), data)?;
                    Ok(buffer)
                })
                .collect::<Result<Vec<_>>>()?,
        )
    };
    let bindings: Vec<_> = buffers.iter().map(Buffer::binding).collect();
    let mut graph = stream.graph()?;
    let mut prior = None;
    for _ in 0..32 {
        prior = Some(unsafe {
            graph.dispatch(
                prior.as_slice(),
                &kernel,
                case.grid,
                case.block,
                &constants,
                &bindings,
            )
        }?);
    }
    let identity = serde_json::json!({"compiler":artifact.compiler_identity(), "artifact":hrx::bundle::digest(artifact.bytes()),
        "report":artifact.report().map(|report| report.json())});
    Ok(Arm {
        graph: graph.finish()?,
        buffers,
        identity,
    })
}

examples/stream_bench.rs (lines 125-132)

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 join(&mut self, after: &[Node]) -> Result<Node>

Record a dependency join with no native payload.

pub fn finish(self) -> Result<GraphExec>

Transfer recorded commands and allocation charges into owned replay state. Large graphs use ordered batches within the native indirect-buffer limit.

Examples found in repository

examples/graph_prepare.rs (line 23)

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/patch_bench.rs (line 320)

fn prepare(
    stream: &mut Stream,
    compiler: &Compiler,
    case: &Case,
    shared: Option<std::sync::Arc<Vec<Buffer>>>,
) -> Result<Arm> {
    let artifact = compiler.module(case.source).compile(&case.spec)?;
    let kernel = unsafe { stream.load_artifact(&artifact) }?;
    let constants = Constants::indices(&kernel, &case.indices)?;
    let buffers = if let Some(buffers) = shared {
        buffers
    } else {
        std::sync::Arc::new(
            case.data
                .iter()
                .map(|data| {
                    let buffer = stream.allocate(data.len())?;
                    stream.upload_blocking(buffer.binding(), data)?;
                    Ok(buffer)
                })
                .collect::<Result<Vec<_>>>()?,
        )
    };
    let bindings: Vec<_> = buffers.iter().map(Buffer::binding).collect();
    let mut graph = stream.graph()?;
    let mut prior = None;
    for _ in 0..32 {
        prior = Some(unsafe {
            graph.dispatch(
                prior.as_slice(),
                &kernel,
                case.grid,
                case.block,
                &constants,
                &bindings,
            )
        }?);
    }
    let identity = serde_json::json!({"compiler":artifact.compiler_identity(), "artifact":hrx::bundle::digest(artifact.bytes()),
        "report":artifact.report().map(|report| report.json())});
    Ok(Arm {
        graph: graph.finish()?,
        buffers,
        identity,
    })
}

examples/stream_bench.rs (line 135)

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 finish_profiled(self, labels: &[String]) -> Result<GraphExec>

Prepare a diagnostic graph with device-clock markers around each payload. Supply one label per non-join node. Instrumentation adds completion barriers and must not be used as end-to-end performance evidence.

Trait Implementations

impl Debug for Graph<'_>

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

Formats the value using the given formatter.