Struct Buffer 

pub struct Buffer { /* private fields */ }

Device backing with allocation-budget ownership.

Implementations

impl Buffer

pub fn bytes(&self) -> usize

Allocation’s exposed byte length.

pub fn binding(&self) -> View<'_>

Whole allocation as a logical binding.

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

examples/patch_bench.rs (line 296)

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,
    })
}
fn check(stream: &mut Stream, arm: &Arm, case: &Case) -> Result<f64> {
    let mut bytes = vec![0; case.expected.len() * 2];
    stream.read_blocking(arm.buffers[case.output].binding(), &mut bytes)?;
    let mut maximum = 0f64;
    for (index, (value, expected)) in bytes
        .as_chunks::<2>()
        .0
        .iter()
        .zip(&case.expected)
        .enumerate()
    {
        let value = float(u16::from_le_bytes(*value));
        let error = (value - expected).abs();
        if !value.is_finite() || error > case.tolerance {
            return Err(hrx::Error::Message(format!(
                "{} output {index}: {value} vs {expected}, error {error}, limit {}",
                case.name, case.tolerance
            )));
        }
        maximum = maximum.max(error);
    }
    Ok(maximum)
}

examples/prepare_weights.rs (line 40)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut args: Vec<_> = std::env::args().skip(1).collect();
    let upload = args.last().is_some_and(|arg| arg == "--upload");
    if upload {
        args.pop();
    }
    if args.len() != 3 || !matches!(args[2].as_str(), "none" | "advice" | "populate") {
        return Err("usage: prepare_weights FILE TENSOR none|advice|populate [--upload]".into());
    }
    // The command requires an immutable checkpoint for the mapping's lifetime.
    let file = unsafe { FileView::map(&args[0]) }?;
    let tensor = file.get(&args[1])?;
    let mut stream = if upload {
        Some(hrx::Stream::open()?)
    } else {
        None
    };
    let buffer = stream
        .as_ref()
        .map(|stream| stream.allocate(tensor.bytes.len()))
        .transpose()?;
    let start = Instant::now();
    let prepared = match args[2].as_str() {
        "populate" => format!("{:?}", file.prepare_bytes(tensor.bytes)?),
        "advice" => {
            file.will_need(tensor);
            "Advised".into()
        }
        _ => "None".into(),
    };
    let preparation_ms = start.elapsed().as_secs_f64() * 1000.;
    let checksum: u64;
    let total_ms;
    if let (Some(stream), Some(buffer)) = (&mut stream, &buffer) {
        stream.upload_blocking(buffer.binding(), tensor.bytes)?;
        total_ms = start.elapsed().as_secs_f64() * 1000.;
        // Validation is outside timing, after the upload fence has retired.
        let mut actual = vec![0; tensor.bytes.len()];
        stream.read_blocking(buffer.binding(), &mut actual)?;
        if actual != tensor.bytes {
            return Err("upload bytes differ from checkpoint".into());
        }
        checksum = actual.iter().map(|&v| u64::from(v)).sum();
    } else {
        checksum = tensor.bytes.iter().map(|&v| u64::from(v)).sum();
        total_ms = start.elapsed().as_secs_f64() * 1000.;
    }
    println!(
        "{}",
        serde_json::json!({"mode":args[2], "preparation":prepared,
        "bytes":tensor.bytes.len(), "preparation_ms":preparation_ms,
        "total_ms":total_ms, "checksum":checksum, "upload":upload,
        "allocation_policy":"preallocated destination; fresh process",
        "timed_scope":if upload { "page preparation plus completed upload" } else { "page preparation plus complete host read" }, "cache_state":"externally controlled"})
    );
    Ok(())
}

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/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 try_slice(&self, offset: usize, length: usize) -> Result<View<'_>>

Checked subrange.

Examples found in repository

examples/stream_bench.rs (line 70)

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 slice(&self, offset: usize, length: usize) -> View<'_>

Checked subrange, panicking on invalid bounds.

pub fn device_ptr(&self) -> Result<*mut c_void>

Mapped host address. Dereferencing requires external synchronization. Ordinary allocations additionally require cache_control before host reads (including partial cache-line updates) and after host writes; allocate_shared omits that requirement.

pub unsafe fn cache_control( &self, flush: bool, offset: usize, length: usize, ) -> Result<()>

Publish host writes (flush = true) or acquire device writes (false). Stream upload/read operations perform these transitions automatically.

Safety

The caller must exclude conflicting host/device access for the entire transition and retain the allocation until all accesses have completed.

Trait Implementations

impl Debug for Buffer

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

Formats the value using the given formatter.

impl Drop for Buffer

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.