Cache materialized spans per context and size emitter extents from the subject query

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<crate::node::List<'a, T>>,
+	total: &'a dyn Fn() -> GPoll<crate::gpoll::Extent>,
 }
 
 impl<'a, T> ListIn<'a, T> {
-	pub fn new(get: &'a dyn Fn() -> GPoll<crate::node::List<'a, T>>) -> Self {
+	pub fn new(get: &'a dyn Fn() -> GPoll<crate::node::List<'a, T>>, total: &'a dyn Fn() -> GPoll<crate::gpoll::Extent>) -> Self {
-		Self { get }
+		Self { get, total }
 	}
 
 	pub fn get(&self) -> GPoll<crate::node::List<'a, T>> {
 		(self.get)()
 	}
+
+	/// The subject wire's total flat extent as a plain query.
+	pub fn total(&self) -> GPoll<crate::gpoll::Extent> {
+		(self.total)()
+	}
 }
 
 /// The queried absolute level (innermost `0`), paired with the node's depth.

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<usize> {
+	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<TokenStream2> {
 	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<fn() -> (#(#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<TokenStream2> = 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) {
+						let __mat_key = {
-							#core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::Exactly(__count)) => ::core::result::Result::Ok(__count),
+							let mut __keyed = *__input;
-							#core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::AtLeast(__bound)) => ::core::result::Result::Err(__bound),
+							#core_types::context::InjectIndex::set_index(&mut __keyed, 0);
-							#core_types::gpoll::GPoll::Pending => #pending,
+							#core_types::registry::cache_key(&__keyed)
-							_ => #non_exact,
 						};
-						let __batch = match __sized {
+						let __mat_generation = __arena.generation();
-							::core::result::Result::Ok(__count) => {
+						let __mat_hit = match *self.#cache_slot.lock().unwrap() {
-								let __start: u64 = 0;
+							::core::option::Option::Some((__key, __generation, __base, __len)) if __key == __mat_key && __generation == __mat_generation => {
-								match #core_types::record::materialize_batch(&self.#name, __input, __start..__start + __count as u64, __arena) {
+								::core::option::Option::Some((__base, __len))
-									#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,
-								}
 							}
-							// The count is a lower bound: drain by guess-and-double
+							_ => ::core::option::Option::None,
-							// until a short fill, each reply's hint seeding the next
+						};
-							// guess.
+						let __batch = match __mat_hit {
-							::core::result::Result::Err(__bound) => {
+							// SAFETY: within the generation the cached batch stays
-								let mut __guess = __bound.max(16);
+							// live, immutable, and of this edge's layout.
-								loop {
+							::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)) },
-									let (__batch, __hint) = match #core_types::record::materialize_batch(&self.#name, __input, 0..__guess as u64, __arena) {
+							::core::option::Option::None => {
-										#core_types::node::BatchStatus::Lent(__batch, _, __hint) => (__batch, __hint),
+								let __sized = match #core_types::node::Node::extent(&self.#name, __input, #core_types::gpoll::Level::Total) {
-										#core_types::node::BatchStatus::Filled(__batch, _, __hint) => (__batch.into_shared(), __hint),
+									#core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::Exactly(__count)) => ::core::result::Result::Ok(__count),
-										#core_types::node::BatchStatus::Pending => #pending,
+									#core_types::gpoll::GPoll::Final(#core_types::gpoll::Extent::AtLeast(__bound)) => ::core::result::Result::Err(__bound),
-										#core_types::node::BatchStatus::Error(__error) => #batch_error,
+									#core_types::gpoll::GPoll::Pending => #pending,
-										_ => #batch_failed,
+									_ => #non_exact,
-									};
+								};
-									let __filled = __batch.len();
+								let __fresh = match __sized {
-									if __filled < __guess {
+									::core::result::Result::Ok(__count) => {
-										break __batch;
+										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 {
+									// The count is a lower bound: drain by guess-and-double
-										#core_types::gpoll::Extent::Exactly(__total) if __total <= __filled => break __batch,
+									// until a short fill, each reply's hint seeding the next
-										#core_types::gpoll::Extent::Exactly(__total) => __guess = __total,
+									// guess.
-										#core_types::gpoll::Extent::AtLeast(__more) => __guess = (__guess * 2).max(__more),
+									::core::result::Result::Err(__bound) => {
-										#core_types::gpoll::Extent::Free => __guess *= 2,
+										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(&regular_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(&regular_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)*
 					}
 				}
 			}

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<T>(legacy: &List<T>, relative_to_bounds: ReferencePoint, offset: f64, angle: f64, keep_original: bool) -> Extent
-where
-	List<T>: 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<T: Clone + Send + Sync + 'static>(content: core_types::node::List<'_, T>) -> List<T> {
 	// 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>,
+	_relative_to_bounds: ValueIn<'_, ReferencePoint>,
-	offset: ValueIn<'_, f64>,
+	_offset: ValueIn<'_, f64>,
-	angle: ValueIn<'_, f64>,
+	_angle: ValueIn<'_, f64>,
 	keep_original: ValueIn<'_, bool>,
 	level: LevelIn,
 ) -> GPoll<Extent> {
 	match level.top() {
-		true => content
+		true => content.total().zip(keep_original.get()).map(|(total, keep_original)| match (total, keep_original) {
-			.get()
+			(total, false) => total,
-			.zip(relative_to_bounds.get())
+			(Extent::Exactly(count), true) => Extent::Exactly(count * 2),
-			.zip(offset.get())
+			(Extent::AtLeast(bound), true) => Extent::AtLeast(bound * 2),
-			.zip(angle.get())
+			(Extent::Free, true) => Extent::Free,
-			.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)
-			}),
 		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>,
+	_relative_to_bounds: ValueIn<'_, ReferencePoint>,
-	offset: ValueIn<'_, f64>,
+	_offset: ValueIn<'_, f64>,
-	angle: ValueIn<'_, f64>,
+	_angle: ValueIn<'_, f64>,
 	keep_original: ValueIn<'_, bool>,
 	level: LevelIn,
 ) -> GPoll<Extent> {
 	match level.top() {
-		true => content
+		true => content.total().zip(keep_original.get()).map(|(total, keep_original)| match (total, keep_original) {
-			.get()
+			(total, false) => total,
-			.zip(relative_to_bounds.get())
+			(Extent::Exactly(count), true) => Extent::Exactly(count * 2),
-			.zip(offset.get())
+			(Extent::AtLeast(bound), true) => Extent::AtLeast(bound * 2),
-			.zip(angle.get())
+			(Extent::Free, true) => Extent::Free,
-			.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)
-			}),
 		false => GPoll::Final(Extent::Exactly(1)),
 	}
 }

node-graph/nodes/vector/src/vector_nodes.rs

@@ -149,7 +149,7 @@ fn assign_colors_extent(
 	level: LevelIn,
 ) -> GPoll<Extent> {
 	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<Extent> {
 	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<T>(
 	Err(GraphError::past_end().into())
 }
 
-/// The pushed level holds one copy per point; inner levels forward to the
+/// The pushed level holds one copy per point (a data-dependent count, so the
-/// content, taken uniform across copies.
+/// 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.
+/// A fill-bearing row splits into a fill lane and a solidified stroke lane,
-fn solidify_extent_of(graphic_list: List<Graphic>) -> Extent {
+/// so the count depends on the content: the level reports the subject's
-	let flattened: List<Vector> = graphic_list.into_flattened_list();
+/// count as a lower bound and consumers drain to the past-end signal.
-	Extent::Exactly((0..flattened.len()).map(|index| 1 + usize::from(has_paint_at(&flattened, index, ATTR_FILL))).sum())
-}
-
 fn solidify_stroke_extent(content: ListIn<'_, Graphic>, level: LevelIn) -> GPoll<Extent> {
 	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<Extent> {
 	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)),
 	}
 }