From 3fb8b10f80fdc3af898ca637adac42de6393c4ef Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Sun, 23 Aug 2026 00:57:37 +0000 Subject: [PATCH] Cache materialized spans per context and size emitter extents from the subject query --- node-graph/libraries/core-types/src/extent.rs | 15 +- node-graph/node-macro/src/codegen.rs | 144 +++++++++++++----- node-graph/nodes/graphic/src/graphic.rs | 62 +++----- node-graph/nodes/vector/src/vector_nodes.rs | 29 ++-- 4 files changed, 157 insertions(+), 93 deletions(-) diff --git a/node-graph/libraries/core-types/src/extent.rs b/node-graph/libraries/core-types/src/extent.rs index 40d06fba11..f067442bed 100644 --- a/node-graph/libraries/core-types/src/extent.rs +++ b/node-graph/libraries/core-types/src/extent.rs @@ -44,19 +44,28 @@ impl<'a> ExtentIn<'a> { /// A ranked (`IList`) input materialized whole: `get` drives the batch and /// yields the level as a [`List`](crate::node::List), for extents that depend -/// on the input's data rather than its counts alone. +/// on the input's data rather than its counts alone. `total` answers the +/// subject's flat count without materializing, so count-shaped extents stay +/// cheap: a materializing extent inside another's subject multiplies, and +/// nested emitters turn that into a blowup. pub struct ListIn<'a, T> { get: &'a dyn Fn() -> GPoll>, + total: &'a dyn Fn() -> GPoll, } impl<'a, T> ListIn<'a, T> { - pub fn new(get: &'a dyn Fn() -> GPoll>) -> Self { - Self { get } + pub fn new(get: &'a dyn Fn() -> GPoll>, total: &'a dyn Fn() -> GPoll) -> Self { + Self { get, total } } pub fn get(&self) -> GPoll> { (self.get)() } + + /// The subject wire's total flat extent as a plain query. + pub fn total(&self) -> GPoll { + (self.total)() + } } /// The queried absolute level (innermost `0`), paired with the node's depth. diff --git a/node-graph/node-macro/src/codegen.rs b/node-graph/node-macro/src/codegen.rs index f7b997659f..47a87362fc 100644 --- a/node-graph/node-macro/src/codegen.rs +++ b/node-graph/node-macro/src/codegen.rs @@ -21,6 +21,17 @@ use metadata::generate_node_input_references; static NODE_ID: AtomicU64 = AtomicU64::new(0); +/// The regular inputs that materialize whole in the eval prologue, which is +/// where the per-node batch cache slots attach. +fn materialized_indices(regular_fields: &[&ParsedField], node: &ir::Node) -> Vec { + regular_fields + .iter() + .enumerate() + .filter(|(index, field)| matches!(field.ty, ParsedFieldType::Regular(_)) && matches!(ir::value_binding(node, *index), ValueBinding::Materialized)) + .map(|(index, _)| index) + .collect() +} + pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn) -> syn::Result { let ParsedNodeFn { attributes, @@ -232,6 +243,13 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn if !carried_generic_idents.is_empty() { record_state_fields.push(quote!(pub(super) __marker: ::core::marker::PhantomData (#(#carried_generic_idents,)*)>)); } + // One slot per materialized input: the batch of the frame that + // materialized it, keyed by (lane-normalized context, generation), so + // per-lane evals share one materialization. + record_state_fields.extend(materialized_indices(&struct_regular_fields, &node).into_iter().map(|index| { + let slot = format_ident!("__mat_cache_{index}"); + quote!(pub(super) #slot: ::std::sync::Arc<::std::sync::Mutex<::core::option::Option<(u64, u64, usize, usize)>>>) + })); let async_source = parsed.injects_async_source_fields(); let slot_value_type = slot_value_type(output_type); @@ -441,6 +459,13 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn }) .into_iter(); let marker_init = (!carried_generic_idents.is_empty()).then(|| quote!(__marker: ::core::marker::PhantomData,)).into_iter(); + let plain_mat_cache_inits: Vec = materialized_indices(&struct_regular_fields, &node) + .into_iter() + .map(|index| { + let slot = format_ident!("__mat_cache_{index}"); + quote!(#slot: ::core::default::Default::default(),) + }) + .collect(); // `new` carries the bounds the erased glue needs at the output type. let new_where = flip .then(|| { @@ -464,6 +489,7 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn #(#flip_read_inits)* #(#flip_output_inits)* #(#marker_init)* + #(#plain_mat_cache_inits)* } } } @@ -1119,55 +1145,85 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn ValueBinding::Carrier => quote!(), ValueBinding::Materialized => { let fn_name = &parsed.fn_name; + let cache_slot = format_ident!("__mat_cache_{index}"); let non_exact = fail(quote!(#core_types::gpoll::GraphError::new(::std::concat!("reduce over a non-exact extent in ", ::std::stringify!(#fn_name))))); let batch_error = fail(quote!(__error)); let batch_failed = fail(quote!(#core_types::gpoll::GraphError::new("reduce batch failed"))); // A fold consumes the whole subject wire: a deeper wire's // total flat span, sized under the evaluation context, so a - // fold inside a pushed level covers that copy's span. + // fold inside a pushed level covers that copy's span. The + // span caches per (lane-normalized context, generation): + // every lane of a per-lane emitter re-enters this bind, and + // without the cache each lane would re-materialize the + // whole subject. quote! { let __arena = #core_types::context::ExtractArena::arena(__input); - let __sized = match #core_types::node::Node::extent(&self.#name, __input, #core_types::gpoll::Level::Total) { - #core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::Exactly(__count)) => ::core::result::Result::Ok(__count), - #core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::AtLeast(__bound)) => ::core::result::Result::Err(__bound), - #core_types::gpoll::GPoll::Pending => #pending, - _ => #non_exact, + let __mat_key = { + let mut __keyed = *__input; + #core_types::context::InjectIndex::set_index(&mut __keyed, 0); + #core_types::registry::cache_key(&__keyed) }; - let __batch = match __sized { - ::core::result::Result::Ok(__count) => { - let __start: u64 = 0; - match #core_types::record::materialize_batch(&self.#name, __input, __start..__start + __count as u64, __arena) { - #core_types::node::BatchStatus::Lent(__batch, ..) => __batch, - #core_types::node::BatchStatus::Filled(__batch, ..) => __batch.into_shared(), - #core_types::node::BatchStatus::Pending => #pending, - #core_types::node::BatchStatus::Error(__error) => #batch_error, - _ => #batch_failed, - } + let __mat_generation = __arena.generation(); + let __mat_hit = match *self.#cache_slot.lock().unwrap() { + ::core::option::Option::Some((__key, __generation, __base, __len)) if __key == __mat_key && __generation == __mat_generation => { + ::core::option::Option::Some((__base, __len)) } - // The count is a lower bound: drain by guess-and-double - // until a short fill, each reply's hint seeding the next - // guess. - ::core::result::Result::Err(__bound) => { - let mut __guess = __bound.max(16); - loop { - let (__batch, __hint) = match #core_types::record::materialize_batch(&self.#name, __input, 0..__guess as u64, __arena) { - #core_types::node::BatchStatus::Lent(__batch, _, __hint) => (__batch, __hint), - #core_types::node::BatchStatus::Filled(__batch, _, __hint) => (__batch.into_shared(), __hint), - #core_types::node::BatchStatus::Pending => #pending, - #core_types::node::BatchStatus::Error(__error) => #batch_error, - _ => #batch_failed, - }; - let __filled = __batch.len(); - if __filled < __guess { - break __batch; + _ => ::core::option::Option::None, + }; + let __batch = match __mat_hit { + // SAFETY: within the generation the cached batch stays + // live, immutable, and of this edge's layout. + ::core::option::Option::Some((__base, __len)) => unsafe { #core_types::node::RecordBatch::new(__base as *const u8, __len, #core_types::node::Node::<#ctx_ident>::layout(&self.#name)) }, + ::core::option::Option::None => { + let __sized = match #core_types::node::Node::extent(&self.#name, __input, #core_types::gpoll::Level::Total) { + #core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::Exactly(__count)) => ::core::result::Result::Ok(__count), + #core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::AtLeast(__bound)) => ::core::result::Result::Err(__bound), + #core_types::gpoll::GPoll::Pending => #pending, + _ => #non_exact, + }; + let __fresh = match __sized { + ::core::result::Result::Ok(__count) => { + let __start: u64 = 0; + match #core_types::record::materialize_batch(&self.#name, __input, __start..__start + __count as u64, __arena) { + #core_types::node::BatchStatus::Lent(__batch, ..) => __batch, + #core_types::node::BatchStatus::Filled(__batch, ..) => __batch.into_shared(), + #core_types::node::BatchStatus::Pending => #pending, + #core_types::node::BatchStatus::Error(__error) => #batch_error, + _ => #batch_failed, + } } - match __hint { - #core_types::gpoll::Extent::Exactly(__total) if __total <= __filled => break __batch, - #core_types::gpoll::Extent::Exactly(__total) => __guess = __total, - #core_types::gpoll::Extent::AtLeast(__more) => __guess = (__guess * 2).max(__more), - #core_types::gpoll::Extent::Free => __guess *= 2, + // The count is a lower bound: drain by guess-and-double + // until a short fill, each reply's hint seeding the next + // guess. + ::core::result::Result::Err(__bound) => { + let mut __guess = __bound.max(16); + loop { + let (__batch, __hint) = match #core_types::record::materialize_batch(&self.#name, __input, 0..__guess as u64, __arena) { + #core_types::node::BatchStatus::Lent(__batch, _, __hint) => (__batch, __hint), + #core_types::node::BatchStatus::Filled(__batch, _, __hint) => (__batch.into_shared(), __hint), + #core_types::node::BatchStatus::Pending => #pending, + #core_types::node::BatchStatus::Error(__error) => #batch_error, + _ => #batch_failed, + }; + let __filled = __batch.len(); + if __filled < __guess { + break __batch; + } + match __hint { + #core_types::gpoll::Extent::Exactly(__total) if __total <= __filled => break __batch, + #core_types::gpoll::Extent::Exactly(__total) => __guess = __total, + #core_types::gpoll::Extent::AtLeast(__more) => __guess = (__guess * 2).max(__more), + #core_types::gpoll::Extent::Free => __guess *= 2, + } + } } - } + }; + let __base = match __fresh.len() { + 0 => 0usize, + _ => __fresh.get(0).rec().ptr() as usize, + }; + *self.#cache_slot.lock().unwrap() = ::core::option::Option::Some((__mat_key, __mat_generation, __base, __fresh.len())); + __fresh } }; let #name = unsafe { #core_types::node::List::<#ty>::new(__batch) }; @@ -1369,7 +1425,8 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn _ => #core_types::gpoll::GPoll::Error(::std::boxed::Box::new(#core_types::gpoll::GraphError::new("extent could not materialize a ranked input"))), } }; - let #arg = #core_types::extent::ListIn::new(&#query); + let __total = || #core_types::node::Node::extent(&self.#name, __input, #core_types::gpoll::Level::Total); + let #arg = #core_types::extent::ListIn::new(&#query, &__total); } } ValueBinding::RecordElement | ValueBinding::ReadingSecondary => { @@ -2082,6 +2139,10 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn _ => None, })); } + // The materialized-span cache keys on the lane-normalized context. + if !materialized_indices(®ular_fields, &node).is_empty() { + bounds.push(quote!(#ctx_ident: #core_types::graphene_hash::CacheHash + #core_types::context::InjectIndex + ::core::marker::Copy)); + } bounds }; @@ -2240,6 +2301,10 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn let plan_default = (!skips_carrier).then(|| quote!(__plan: ::std::vec::Vec::new(),)).into_iter(); let read_names = (0..flat_reads.len()).map(|index| format_ident!("__read_{index}")).map(|slot| quote!(#slot,)); let write_defaults = (0..write_markers.len()).map(|index| format_ident!("__write_{index}")).map(|slot| quote!(#slot: 0,)); + let mat_cache_defaults = materialized_indices(®ular_fields, &node).into_iter().map(|index| { + let slot = format_ident!("__mat_cache_{index}"); + quote!(#slot: ::core::default::Default::default(),) + }); quote! { #layout_def #layout_meta_def @@ -2259,6 +2324,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn __frame_bytes: 0, #(#read_names)* #(#write_defaults)* + #(#mat_cache_defaults)* } } } diff --git a/node-graph/nodes/graphic/src/graphic.rs b/node-graph/nodes/graphic/src/graphic.rs index 38bb46bc80..8bdaf5367b 100644 --- a/node-graph/nodes/graphic/src/graphic.rs +++ b/node-graph/nodes/graphic/src/graphic.rs @@ -193,9 +193,11 @@ where { let count = legacy.len(); let reflected_transform = mirror_reflection(&legacy, relative_to_bounds, offset, angle); - let (source, mirrored) = match (reflected_transform.is_some() && keep_original, lane < count) { + // Kept originals always double the level so the count stays structural; + // without a reflection (no rectangular bounds) the second half duplicates. + let (source, mirrored) = match (keep_original, lane < count) { (true, true) => (lane, false), - (true, false) => (lane - count, true), + (true, false) => (lane - count, reflected_transform.is_some()), (false, _) => (lane, reflected_transform.is_some()), }; if source >= count { @@ -241,18 +243,6 @@ where )) } -/// The mirrored level: the original lanes (when kept) followed by the -/// reflected lanes. -fn mirror_extent_of(legacy: &List, relative_to_bounds: ReferencePoint, offset: f64, angle: f64, keep_original: bool) -> Extent -where - List: BoundingBox, -{ - match mirror_reflection(legacy, relative_to_bounds, offset, angle) { - Some(_) if keep_original => Extent::Exactly(legacy.len() * 2), - _ => Extent::Exactly(legacy.len()), - } -} - /// The materialized level as its legacy render list. fn legacy_render_list_of(content: core_types::node::List<'_, T>) -> List { // SAFETY: a materialized input's frames are arena-resident. @@ -287,24 +277,23 @@ fn mirror<'e>( mirror_lane(ctx.arena(), legacy_render_list_of(content), ctx.innermost_index() as usize, relative_to_bounds, offset, angle, keep_original) } +/// The kept originals double the level, counted from the subject's extent +/// query alone so nested extents stay materialization-free. fn mirror_extent( content: ListIn<'_, Graphic>, - relative_to_bounds: ValueIn<'_, ReferencePoint>, - offset: ValueIn<'_, f64>, - angle: ValueIn<'_, f64>, + _relative_to_bounds: ValueIn<'_, ReferencePoint>, + _offset: ValueIn<'_, f64>, + _angle: ValueIn<'_, f64>, keep_original: ValueIn<'_, bool>, level: LevelIn, ) -> GPoll { match level.top() { - true => content - .get() - .zip(relative_to_bounds.get()) - .zip(offset.get()) - .zip(angle.get()) - .zip(keep_original.get()) - .map(|((((content, relative_to_bounds), offset), angle), keep_original)| { - mirror_extent_of(&legacy_render_list_of(content), relative_to_bounds, offset, angle, keep_original) - }), + true => content.total().zip(keep_original.get()).map(|(total, keep_original)| match (total, keep_original) { + (total, false) => total, + (Extent::Exactly(count), true) => Extent::Exactly(count * 2), + (Extent::AtLeast(bound), true) => Extent::AtLeast(bound * 2), + (Extent::Free, true) => Extent::Free, + }), false => GPoll::Final(Extent::Exactly(1)), } } @@ -340,22 +329,19 @@ fn mirror_vector<'e>( fn mirror_vector_extent( content: ListIn<'_, Vector>, - relative_to_bounds: ValueIn<'_, ReferencePoint>, - offset: ValueIn<'_, f64>, - angle: ValueIn<'_, f64>, + _relative_to_bounds: ValueIn<'_, ReferencePoint>, + _offset: ValueIn<'_, f64>, + _angle: ValueIn<'_, f64>, keep_original: ValueIn<'_, bool>, level: LevelIn, ) -> GPoll { match level.top() { - true => content - .get() - .zip(relative_to_bounds.get()) - .zip(offset.get()) - .zip(angle.get()) - .zip(keep_original.get()) - .map(|((((content, relative_to_bounds), offset), angle), keep_original)| { - mirror_extent_of(&legacy_render_list_of(content), relative_to_bounds, offset, angle, keep_original) - }), + true => content.total().zip(keep_original.get()).map(|(total, keep_original)| match (total, keep_original) { + (total, false) => total, + (Extent::Exactly(count), true) => Extent::Exactly(count * 2), + (Extent::AtLeast(bound), true) => Extent::AtLeast(bound * 2), + (Extent::Free, true) => Extent::Free, + }), false => GPoll::Final(Extent::Exactly(1)), } } diff --git a/node-graph/nodes/vector/src/vector_nodes.rs b/node-graph/nodes/vector/src/vector_nodes.rs index fcbaeb0714..92330e544f 100644 --- a/node-graph/nodes/vector/src/vector_nodes.rs +++ b/node-graph/nodes/vector/src/vector_nodes.rs @@ -149,7 +149,7 @@ fn assign_colors_extent( level: LevelIn, ) -> GPoll { match level.top() { - true => content.get().map(|content| Extent::Exactly(content.len())), + true => content.total(), false => GPoll::Final(Extent::Exactly(1)), } } @@ -215,7 +215,6 @@ fn assign_colors_graphic<'e>( fn assign_colors_graphic_extent( content: ListIn<'_, Graphic>, - _fill: ValueIn<'_, bool>, _stroke: ValueIn<'_, bool>, _gradient: ListIn<'_, GradientStops>, @@ -226,7 +225,7 @@ fn assign_colors_graphic_extent( level: LevelIn, ) -> GPoll { match level.top() { - true => content.get().map(|content| Extent::Exactly(content.len())), + true => content.total(), false => GPoll::Final(Extent::Exactly(1)), } } @@ -546,8 +545,9 @@ fn copy_to_points( Err(GraphError::past_end().into()) } -/// The pushed level holds one copy per point; inner levels forward to the -/// content, taken uniform across copies. +/// The pushed level holds one copy per point (a data-dependent count, so the +/// points level materializes here; its cone stays small); inner levels +/// forward to the content, taken uniform across copies. fn copy_to_points_extent( content: ExtentIn<'_>, points: ListIn<'_, Vector>, @@ -1605,15 +1605,15 @@ fn solidify_stroke<'e>( solidify_lane(ctx.arena(), legacy_graphic_list_of(content), ctx.innermost_index() as usize) } -/// A fill-bearing row splits into a fill lane and a solidified stroke lane. -fn solidify_extent_of(graphic_list: List) -> Extent { - let flattened: List = graphic_list.into_flattened_list(); - Extent::Exactly((0..flattened.len()).map(|index| 1 + usize::from(has_paint_at(&flattened, index, ATTR_FILL))).sum()) -} - +/// A fill-bearing row splits into a fill lane and a solidified stroke lane, +/// so the count depends on the content: the level reports the subject's +/// count as a lower bound and consumers drain to the past-end signal. fn solidify_stroke_extent(content: ListIn<'_, Graphic>, level: LevelIn) -> GPoll { match level.top() { - true => content.get().map(|content| solidify_extent_of(legacy_graphic_list_of(content))), + true => content.total().map(|total| Extent::AtLeast(match total { + Extent::Exactly(count) | Extent::AtLeast(count) => count, + Extent::Free => 0, + })), false => GPoll::Final(Extent::Exactly(1)), } } @@ -1644,7 +1644,10 @@ fn solidify_stroke_vector<'e>( fn solidify_stroke_vector_extent(content: ListIn<'_, Vector>, level: LevelIn) -> GPoll { match level.top() { - true => content.get().map(|content| solidify_extent_of(legacy_graphic_list_of(content))), + true => content.total().map(|total| Extent::AtLeast(match total { + Extent::Exactly(count) | Extent::AtLeast(count) => count, + Extent::Free => 0, + })), false => GPoll::Final(Extent::Exactly(1)), } }