pub struct Graph<'a> { /* private fields */ }Expand description
A recording of GPU work as a dependency graph, replayed as a unit.
Every operation states what it comes after. Nothing is implicit: &[] records
work that may run as soon as the graph starts, and &[a, b] records work that
waits for both. The runtime schedules the result — graph_analysis.c does a
topological sort, partitions it, and detects independent workstreams, running
up to eight concurrently — so declaring only the edges that exist is what lets
it overlap anything.
Dependencies affect barriers and partitioning; there is no fixed cost per edge. The pinned scheduler considers additional workstreams only after the first 16 recordable nodes of a partition, and an empty join node ends that partition. Declaring branches permits overlap but does not guarantee it.
A dependency can only name an already-recorded node, so a recording is
acyclic by construction and every edge points forward. That also keeps
instantiation on the runtime’s linear fast path. Naming the same node twice
in one list is collapsed rather than rejected, so after may be assembled
from overlapping stage outputs.
Addresses, constants and shapes are fixed at record time. Native capture and
graph-exec update are unimplemented in the pinned revision rather than merely
unwrapped: hrx_graph_exec_update is a 17-byte stub and
hrx_stream_capture_status is 3 bytes, so changing a recording means
recording a new one.
Implementations§
Source§impl<'a> Graph<'a>
impl<'a> Graph<'a>
Sourcepub fn fill(
&mut self,
after: &[Node],
dst: View<'a>,
pattern: u8,
) -> Result<Node>
pub fn fill( &mut self, after: &[Node], dst: View<'a>, pattern: u8, ) -> Result<Node>
Record a byte-pattern fill over a nonempty span on this device.
Examples found in repository?
examples/graph_prepare.rs (line 21)
4fn main() -> hrx::Result<()> {
5 let runtime = Runtime::new()?;
6 let buffer = runtime.allocate(4096, MemoryPlacement::GpuLocal)?;
7 let stream = Stream::open()?;
8 let raw = stream.allocate(4096)?;
9 for count in [128, 512, 1024, 2048] {
10 let begin = Instant::now();
11 let mut graph = runtime.graph();
12 for _ in 0..count {
13 graph.fill(buffer.view(), 0)?;
14 }
15 let graph = graph.prepare()?;
16 let inferred = begin.elapsed();
17 let begin = Instant::now();
18 let mut native = stream.graph()?;
19 let mut last = None;
20 for _ in 0..count {
21 last = Some(native.fill(last.as_slice(), raw.binding(), 0)?);
22 }
23 let _native = native.finish()?;
24 println!(
25 "nodes={count} inferred_us={} explicit_us={}",
26 inferred.as_micros(),
27 begin.elapsed().as_micros()
28 );
29 graph.submit()?.wait()?;
30 }
31 Ok(())
32}Sourcepub fn copy(
&mut self,
after: &[Node],
dst: View<'a>,
src: View<'a>,
) -> Result<Node>
pub fn copy( &mut self, after: &[Node], dst: View<'a>, src: View<'a>, ) -> Result<Node>
Record a copy between equal, nonempty spans on this device.
Sourcepub unsafe fn dispatch(
&mut self,
after: &[Node],
kernel: &'a Kernel,
grid: [u32; 3],
block: [u32; 3],
constants: &Constants,
bindings: &[View<'a>],
) -> Result<Node>
pub unsafe fn dispatch( &mut self, after: &[Node], kernel: &'a Kernel, grid: [u32; 3], block: [u32; 3], constants: &Constants, bindings: &[View<'a>], ) -> Result<Node>
Record a kernel invocation.
§Safety
As Stream::dispatch. Constants, addresses and grid are fixed for every
replay. Nodes that access overlapping spans, with at least one write,
must be ordered through after. Shared read-only weights need no edge.
Examples found in repository?
examples/stream_bench.rs (lines 125-132)
30fn main() -> Result<()> {
31 let mut stream = Stream::open()?;
32 let mut metrics = BTreeMap::new();
33 let samples = std::env::var("HRX_BENCH_SAMPLES")
34 .ok()
35 .and_then(|value| value.parse::<usize>().ok())
36 .filter(|&value| value > 0)
37 .unwrap_or(9);
38 let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
39 for _ in 0..50_000 {
40 let buffer = stream.scratch(1024 * 1024)?;
41 black_box(&buffer);
42 stream.recycle(buffer)?;
43 }
44 Ok(())
45 })?;
46 metrics.insert("scratch_reuse_ns", ns);
47 let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
48 for _ in 0..2_000 {
49 black_box(stream.allocate(1024 * 1024)?);
50 }
51 Ok(())
52 })?;
53 metrics.insert("allocate_drop_ns", ns);
54
55 let bytes = 256 * 1024 * 1024;
56 let chunk = vec![0x3cu8; 4 * 1024 * 1024];
57 let weights = stream.allocate(bytes)?;
58 let (_, ns) = measure(&mut stream, samples, 1, |stream| {
59 for offset in (0..bytes).step_by(chunk.len()) {
60 stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
61 }
62 Ok(())
63 })?;
64 metrics.insert(
65 "queued_upload_gib_s",
66 bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
67 );
68 let mut check = [0u8; 16];
69 stream.read_blocking(
70 weights.try_slice(bytes - check.len(), check.len())?,
71 &mut check,
72 )?;
73 assert_eq!(check, [0x3c; 16]);
74 drop(weights);
75
76 let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
77 let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
78 let mut spec = hrx::loom::Specialization::new("krea2_euler");
79 spec.set_config("krea2.euler.grid_x", "1");
80 spec.set_config("krea2.euler.grid_y", "1");
81 let artifact = module.compile(&spec)?;
82 // This repository owns the source; use its exact dimensions and scalar layout.
83 let kernel = unsafe { stream.load_artifact(&artifact)? };
84 let sample = stream.allocate(512)?;
85 let velocity = stream.allocate(512)?;
86 let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
87 stream.upload_blocking(sample.binding(), &ones)?;
88 stream.upload_blocking(velocity.binding(), &ones)?;
89 let mut constants = Constants::new();
90 match kernel.info().constant_byte_length {
91 8 => constants.push(256u32)?,
92 12 => constants.push(256u64)?,
93 n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
94 }
95 constants.push(0f32)?;
96 let bindings = [sample.binding(), velocity.binding()];
97 let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
98 for _ in 0..2048 {
99 // The source accesses exactly the 256 bf16 elements in each buffer.
100 unsafe {
101 stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
102 }
103 }
104 Ok(())
105 })?;
106 metrics.insert("dispatch_enqueue_ns", enqueue);
107 metrics.insert("dispatch_complete_ns", complete);
108
109 // Use disjoint allocations in both graphs. Even an idempotent store is a
110 // write: concurrent Euler dispatches cannot share their in/out buffer.
111 let mut graph_buffers = Vec::new();
112 for _ in 0..32 {
113 let sample = stream.allocate(512)?;
114 let velocity = stream.allocate(512)?;
115 stream.upload(sample.binding(), &ones)?;
116 stream.upload(velocity.binding(), &ones)?;
117 graph_buffers.push((sample, velocity));
118 }
119 stream.synchronize()?;
120 let mut builder = stream.graph()?;
121 let mut previous = None;
122 for (sample, velocity) in &graph_buffers {
123 let after = previous.as_slice();
124 previous = Some(unsafe {
125 builder.dispatch(
126 after,
127 &kernel,
128 [1; 3],
129 [256, 1, 1],
130 &constants,
131 &[sample.binding(), velocity.binding()],
132 )?
133 });
134 }
135 let chained = builder.finish()?;
136
137 // The same disjoint workloads, this time allowing concurrent scheduling.
138 let mut builder = stream.graph()?;
139 for (sample, velocity) in &graph_buffers {
140 unsafe {
141 builder.dispatch(
142 &[],
143 &kernel,
144 [1; 3],
145 [256, 1, 1],
146 &constants,
147 &[sample.binding(), velocity.binding()],
148 )?;
149 }
150 }
151 let independent = builder.finish()?;
152 let mut graphs = [chained, independent];
153 let mut enqueue = [Vec::new(), Vec::new()];
154 let mut complete = [Vec::new(), Vec::new()];
155 for round in 0..samples + 3 {
156 for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
157 stream.synchronize()?;
158 let start = Instant::now();
159 for _ in 0..256 {
160 stream.launch(&mut graphs[arm])?;
161 }
162 let submitted = start.elapsed();
163 stream.synchronize()?;
164 let elapsed = start.elapsed();
165 if round >= 3 {
166 enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
167 complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
168 }
169 }
170 }
171 for values in enqueue.iter_mut().chain(complete.iter_mut()) {
172 values.sort_by(f64::total_cmp);
173 }
174 let ns = complete[0][samples / 2];
175 let ns_free = complete[1][samples / 2];
176 metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
177 metrics.insert("graph_complete_ns_per_kernel", ns);
178 metrics.insert("graph_independent_ns_per_kernel", ns_free);
179 metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
180 for (sample, _) in &graph_buffers {
181 let mut output = vec![0; 512];
182 stream.read_blocking(sample.binding(), &mut output)?;
183 assert_eq!(output, ones);
184 }
185 let mut output = vec![0; 512];
186 stream.read_blocking(sample.binding(), &mut output)?;
187 assert_eq!(output, ones);
188 println!("{}", serde_json::to_string(&metrics)?);
189 Ok(())
190}Sourcepub fn join(&mut self, after: &[Node]) -> Result<Node>
pub fn join(&mut self, after: &[Node]) -> Result<Node>
Record a node that does no work and exists only to collect dependencies.
The pinned runtime gives this node its own partition and queue barrier. For one consumer, pass the producer nodes directly to that operation. A join can express a shared dependency, but is not necessarily cheaper.
Sourcepub fn finish(self) -> Result<GraphExec>
pub fn finish(self) -> Result<GraphExec>
Instantiate the recording, retaining native resources independently of its borrows. A cyclic graph is rejected here.
Examples found in repository?
examples/graph_prepare.rs (line 23)
4fn main() -> hrx::Result<()> {
5 let runtime = Runtime::new()?;
6 let buffer = runtime.allocate(4096, MemoryPlacement::GpuLocal)?;
7 let stream = Stream::open()?;
8 let raw = stream.allocate(4096)?;
9 for count in [128, 512, 1024, 2048] {
10 let begin = Instant::now();
11 let mut graph = runtime.graph();
12 for _ in 0..count {
13 graph.fill(buffer.view(), 0)?;
14 }
15 let graph = graph.prepare()?;
16 let inferred = begin.elapsed();
17 let begin = Instant::now();
18 let mut native = stream.graph()?;
19 let mut last = None;
20 for _ in 0..count {
21 last = Some(native.fill(last.as_slice(), raw.binding(), 0)?);
22 }
23 let _native = native.finish()?;
24 println!(
25 "nodes={count} inferred_us={} explicit_us={}",
26 inferred.as_micros(),
27 begin.elapsed().as_micros()
28 );
29 graph.submit()?.wait()?;
30 }
31 Ok(())
32}More examples
examples/stream_bench.rs (line 135)
30fn main() -> Result<()> {
31 let mut stream = Stream::open()?;
32 let mut metrics = BTreeMap::new();
33 let samples = std::env::var("HRX_BENCH_SAMPLES")
34 .ok()
35 .and_then(|value| value.parse::<usize>().ok())
36 .filter(|&value| value > 0)
37 .unwrap_or(9);
38 let (ns, _) = measure(&mut stream, samples, 50_000, |stream| {
39 for _ in 0..50_000 {
40 let buffer = stream.scratch(1024 * 1024)?;
41 black_box(&buffer);
42 stream.recycle(buffer)?;
43 }
44 Ok(())
45 })?;
46 metrics.insert("scratch_reuse_ns", ns);
47 let (ns, _) = measure(&mut stream, samples, 2_000, |stream| {
48 for _ in 0..2_000 {
49 black_box(stream.allocate(1024 * 1024)?);
50 }
51 Ok(())
52 })?;
53 metrics.insert("allocate_drop_ns", ns);
54
55 let bytes = 256 * 1024 * 1024;
56 let chunk = vec![0x3cu8; 4 * 1024 * 1024];
57 let weights = stream.allocate(bytes)?;
58 let (_, ns) = measure(&mut stream, samples, 1, |stream| {
59 for offset in (0..bytes).step_by(chunk.len()) {
60 stream.upload(weights.binding().slice(offset, chunk.len())?, &chunk)?;
61 }
62 Ok(())
63 })?;
64 metrics.insert(
65 "queued_upload_gib_s",
66 bytes as f64 / (1024f64.powi(3) * ns * 1e-9),
67 );
68 let mut check = [0u8; 16];
69 stream.read_blocking(
70 weights.try_slice(bytes - check.len(), check.len())?,
71 &mut check,
72 )?;
73 assert_eq!(check, [0x3c; 16]);
74 drop(weights);
75
76 let compiler = hrx::loom::Compiler::for_stream(None, &stream)?;
77 let module = compiler.module(include_str!("../tests/kernels/euler.loom"));
78 let mut spec = hrx::loom::Specialization::new("krea2_euler");
79 spec.set_config("krea2.euler.grid_x", "1");
80 spec.set_config("krea2.euler.grid_y", "1");
81 let artifact = module.compile(&spec)?;
82 // This repository owns the source; use its exact dimensions and scalar layout.
83 let kernel = unsafe { stream.load_artifact(&artifact)? };
84 let sample = stream.allocate(512)?;
85 let velocity = stream.allocate(512)?;
86 let ones: Vec<u8> = (0..256).flat_map(|_| 0x3f80u16.to_le_bytes()).collect();
87 stream.upload_blocking(sample.binding(), &ones)?;
88 stream.upload_blocking(velocity.binding(), &ones)?;
89 let mut constants = Constants::new();
90 match kernel.info().constant_byte_length {
91 8 => constants.push(256u32)?,
92 12 => constants.push(256u64)?,
93 n => return Err(hrx::Error::Message(format!("unexpected constant size {n}"))),
94 }
95 constants.push(0f32)?;
96 let bindings = [sample.binding(), velocity.binding()];
97 let (enqueue, complete) = measure(&mut stream, samples, 2048, |stream| {
98 for _ in 0..2048 {
99 // The source accesses exactly the 256 bf16 elements in each buffer.
100 unsafe {
101 stream.dispatch(&kernel, [1; 3], [256, 1, 1], &constants, &bindings)?;
102 }
103 }
104 Ok(())
105 })?;
106 metrics.insert("dispatch_enqueue_ns", enqueue);
107 metrics.insert("dispatch_complete_ns", complete);
108
109 // Use disjoint allocations in both graphs. Even an idempotent store is a
110 // write: concurrent Euler dispatches cannot share their in/out buffer.
111 let mut graph_buffers = Vec::new();
112 for _ in 0..32 {
113 let sample = stream.allocate(512)?;
114 let velocity = stream.allocate(512)?;
115 stream.upload(sample.binding(), &ones)?;
116 stream.upload(velocity.binding(), &ones)?;
117 graph_buffers.push((sample, velocity));
118 }
119 stream.synchronize()?;
120 let mut builder = stream.graph()?;
121 let mut previous = None;
122 for (sample, velocity) in &graph_buffers {
123 let after = previous.as_slice();
124 previous = Some(unsafe {
125 builder.dispatch(
126 after,
127 &kernel,
128 [1; 3],
129 [256, 1, 1],
130 &constants,
131 &[sample.binding(), velocity.binding()],
132 )?
133 });
134 }
135 let chained = builder.finish()?;
136
137 // The same disjoint workloads, this time allowing concurrent scheduling.
138 let mut builder = stream.graph()?;
139 for (sample, velocity) in &graph_buffers {
140 unsafe {
141 builder.dispatch(
142 &[],
143 &kernel,
144 [1; 3],
145 [256, 1, 1],
146 &constants,
147 &[sample.binding(), velocity.binding()],
148 )?;
149 }
150 }
151 let independent = builder.finish()?;
152 let mut graphs = [chained, independent];
153 let mut enqueue = [Vec::new(), Vec::new()];
154 let mut complete = [Vec::new(), Vec::new()];
155 for round in 0..samples + 3 {
156 for arm in if round % 2 == 0 { [0, 1] } else { [1, 0] } {
157 stream.synchronize()?;
158 let start = Instant::now();
159 for _ in 0..256 {
160 stream.launch(&mut graphs[arm])?;
161 }
162 let submitted = start.elapsed();
163 stream.synchronize()?;
164 let elapsed = start.elapsed();
165 if round >= 3 {
166 enqueue[arm].push(submitted.as_secs_f64() * 1e9 / 256.0);
167 complete[arm].push(elapsed.as_secs_f64() * 1e9 / (256.0 * 32.0));
168 }
169 }
170 }
171 for values in enqueue.iter_mut().chain(complete.iter_mut()) {
172 values.sort_by(f64::total_cmp);
173 }
174 let ns = complete[0][samples / 2];
175 let ns_free = complete[1][samples / 2];
176 metrics.insert("graph_enqueue_ns_per_replay", enqueue[0][samples / 2]);
177 metrics.insert("graph_complete_ns_per_kernel", ns);
178 metrics.insert("graph_independent_ns_per_kernel", ns_free);
179 metrics.insert("graph_scheduling_difference_ns_per_kernel", ns - ns_free);
180 for (sample, _) in &graph_buffers {
181 let mut output = vec![0; 512];
182 stream.read_blocking(sample.binding(), &mut output)?;
183 assert_eq!(output, ones);
184 }
185 let mut output = vec![0; 512];
186 stream.read_blocking(sample.binding(), &mut output)?;
187 assert_eq!(output, ones);
188 println!("{}", serde_json::to_string(&metrics)?);
189 Ok(())
190}Trait Implementations§
Auto Trait Implementations§
impl<'a> !Send for Graph<'a>
impl<'a> !Sync for Graph<'a>
impl<'a> Freeze for Graph<'a>
impl<'a> RefUnwindSafe for Graph<'a>
impl<'a> Unpin for Graph<'a>
impl<'a> UnsafeUnpin for Graph<'a>
impl<'a> UnwindSafe for Graph<'a>
Blanket Implementations§
Source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
Source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
Mutably borrows from an owned value. Read more
Source§impl<T> DropFlavorWrapper<T> for T
impl<T> DropFlavorWrapper<T> for T
Source§impl<T, W> HasTypeWitness<W> for Twhere
W: MakeTypeWitness<Arg = T>,
T: ?Sized,
impl<T, W> HasTypeWitness<W> for Twhere
W: MakeTypeWitness<Arg = T>,
T: ?Sized,
Source§impl<T> Identity for Twhere
T: ?Sized,
impl<T> Identity for Twhere
T: ?Sized,
Source§impl<T> Instrument for T
impl<T> Instrument for T
Source§fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
Source§fn in_current_span(self) -> Instrumented<Self> ⓘ
fn in_current_span(self) -> Instrumented<Self> ⓘ
Source§impl<T> IntoEither for T
impl<T> IntoEither for T
Source§fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
Converts
self into a Left variant of Either<Self, Self>
if into_left is true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read moreSource§fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
Converts
self into a Left variant of Either<Self, Self>
if into_left(&self) returns true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read more