From aed78550022febb720b7482a8a0afa81ecc53e43 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Fri, 11 Sep 2026 14:10:56 +0000 Subject: [PATCH] Add a map_element primitive to Lane and teach the node macro to emit it --- node-graph/libraries/core-types/src/node.rs | 40 ++++++- .../libraries/core-types/src/record/serve.rs | 17 +++ node-graph/node-macro/src/codegen.rs | 43 +++++-- node-graph/nodes/gcore/src/record.rs | 110 +++++++++++++++++- 4 files changed, 193 insertions(+), 17 deletions(-) diff --git a/node-graph/libraries/core-types/src/node.rs b/node-graph/libraries/core-types/src/node.rs index 6c4c8bdd19..dda3a12723 100644 --- a/node-graph/libraries/core-types/src/node.rs +++ b/node-graph/libraries/core-types/src/node.rs @@ -188,20 +188,48 @@ impl<'a> RecordLane<'a> { /// One lane of a materialized level, element-typed. In a kernel's element /// position the output frame is copied from this lane. +/// +/// The subject is substitutable: [`map_element`](Self::map_element) keeps the +/// lane's whole record and swaps what it is a record *of*, so a node that +/// rewrites its element still carries every column it never mentions. `Attr` +/// members of the kernel's return tuple remain overrides layered on top. #[derive(Debug)] pub struct Lane<'a, T> { lane: RecordLane<'a>, - _element: PhantomData, + /// `Some` once the subject was replaced; the carried record's own element + /// is then ignored in favour of this value. + subject: Option, } -// A view regardless of `T`: copying a lane copies no record. -impl Clone for Lane<'_, T> { +// A lane that carries no substitute is still a pure view: copying it copies no +// record. One holding a replaced subject owns it, so it clones with the value. +impl Clone for Lane<'_, T> { fn clone(&self) -> Self { - *self + Self { + lane: self.lane, + subject: self.subject.clone(), + } } } -impl Copy for Lane<'_, T> {} +impl Copy for Lane<'_, T> {} + +impl<'a, T> Lane<'a, T> { + /// The same source lane with a different subject. The record's columns are + /// carried unchanged; only the element becomes `element`. + pub fn map_element(self, element: U) -> Lane<'a, U> { + Lane { + lane: self.lane, + subject: Some(element), + } + } + + /// The substituted subject, or `None` where this lane still stands for its + /// source record's own element. + pub fn into_element(self) -> Option { + self.subject + } +} impl<'a, T> Deref for Lane<'a, T> { type Target = RecordLane<'a>; @@ -276,7 +304,7 @@ impl<'a, T> List<'a, T> { pub fn lane(&self, index: usize) -> Lane<'a, T> { Lane { lane: self.batch.get(index), - _element: PhantomData, + subject: None, } } diff --git a/node-graph/libraries/core-types/src/record/serve.rs b/node-graph/libraries/core-types/src/record/serve.rs index bafb4487c1..6b2ba5c779 100644 --- a/node-graph/libraries/core-types/src/record/serve.rs +++ b/node-graph/libraries/core-types/src/record/serve.rs @@ -127,6 +127,23 @@ impl<'e, 'l> FrameClaim<'e, 'l> { self.filled_fields = true; } + /// Copies the source record's element bytes into the frame, for a gathered + /// lane whose subject [`map_element`](crate::node::Lane::map_element) never + /// replaced. Carrying the element is not a field write, so it leaves + /// `filled_fields` alone. + /// + /// # Safety + /// As [`Self::carry`]: `src` must be a live record of the gather's source + /// layout and must not overlap this frame. The element slots agree because + /// an unsubstituted `Lane` can only come from a level whose element is + /// already `T`. + pub unsafe fn carry_element(&mut self, src: Rec<'_>) { + let size = self.layout.element.size; + if size > 0 { + unsafe { apply_plan(src, self.dst(), &[(0, 0, size)]) }; + } + } + /// Writes a field at its wiring-resolved offset. /// /// # Safety diff --git a/node-graph/node-macro/src/codegen.rs b/node-graph/node-macro/src/codegen.rs index 3f2d420f1f..64aa2b8ea8 100644 --- a/node-graph/node-macro/src/codegen.rs +++ b/node-graph/node-macro/src/codegen.rs @@ -822,9 +822,13 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn // The record-io write set, resolved from the output item and carrier input. let write_markers: Vec<&Type> = node.output.shape.attrs.iter().map(|attr| &attr.marker).collect(); let removes: Vec<&Type> = node.output.removes.iter().map(|attr| &attr.marker).collect(); - // A gathered element rides the copy plan, never a write. + // A gathered element of a known type is a write, not a plan entry: the lane's + // subject is substitutable through `map_element`, so the frame takes the + // replacement where there is one and carries the source bytes where there is + // not. A gathered element that is still generic keeps the plan's byte carry + // and cannot be substituted. let element_write: Option<&Type> = match &node.output.shape.element { - crate::codegen::ir::Element::Concrete(ty) if !gather_carrier => Some(ty), + crate::codegen::ir::Element::Concrete(ty) => Some(ty), _ => None, }; let carrier_read_ty: Option = node.inputs.first().filter(|input| input.subject).and_then(|input| match &input.shape.element { @@ -2120,14 +2124,33 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn // A droppable element parks in the arena and rides as a reference. let element_store = element_write.map(|ty| { let ty = &crate::codegen::classify::substitute_lifetimes(ty, "'_"); - quote! { - if __frame.element::<#ty>(__element, #core_types::context::ExtractArena::arena(__input)).is_none() { - return #core_types::gpoll::Interrupt::from(#core_types::gpoll::GraphError { - kind: #core_types::gpoll::ErrorKind::ArenaExhausted, - trace: ::std::vec::Vec::new(), - }) - .into(); - } + let exhausted = quote! { + return #core_types::gpoll::Interrupt::from(#core_types::gpoll::GraphError { + kind: #core_types::gpoll::ErrorKind::ArenaExhausted, + trace: ::std::vec::Vec::new(), + }) + .into(); + }; + match gather_carrier { + // A gathered lane writes the subject `map_element` substituted, + // and otherwise carries the source record's own element bytes. + true => quote! { + match #core_types::node::Lane::into_element(__element) { + ::core::option::Option::Some(__subject) => { + if __frame.element::<#ty>(__subject, #core_types::context::ExtractArena::arena(__input)).is_none() { + #exhausted + } + } + // SAFETY: `__src_rec` is the gathered lane's own live record, whose + // element slot the output layout resolved against. + ::core::option::Option::None => unsafe { __frame.carry_element(__src_rec) }, + } + }, + false => quote! { + if __frame.element::<#ty>(__element, #core_types::context::ExtractArena::arena(__input)).is_none() { + #exhausted + } + }, } }); let attr_stores = attr_binders.iter().enumerate().map(|(index, binder)| { diff --git a/node-graph/nodes/gcore/src/record.rs b/node-graph/nodes/gcore/src/record.rs index 92d688be54..51c58970db 100644 --- a/node-graph/nodes/gcore/src/record.rs +++ b/node-graph/nodes/gcore/src/record.rs @@ -300,7 +300,11 @@ fn mirror_extent(content: ListIn<'_, f64>, keep_original: ValueIn<'_, bool>, lev /// copies that lane's whole record, so undeclared attributes ride along and /// only the declared opacity is rewritten. #[node_macro::node(category("Test"), extent(reverse_lanes_extent))] -fn reverse_lanes(ctx: impl Ctx + ExtractIndex + InjectIndex + Copy, content: IList, opacity: f64) -> Result, Attr)>, Interrupt> { +fn reverse_lanes<'e>( + ctx: impl Ctx + core_types::context::ExtractArena<'e> + ExtractIndex + InjectIndex + Copy, + content: IList, + opacity: f64, +) -> Result, Attr)>, Interrupt> { let lane = ctx.innermost_index() as usize; if lane >= content.len() { return Err(GraphError::past_end().into()); @@ -315,6 +319,24 @@ fn reverse_lanes_extent(content: ListIn<'_, f64>, _opacity: ValueIn<'_, f64>, le } } +/// Gather-carrier kernel with a substituted subject: the lane's record carries +/// exactly as it does for [`reverse_lanes`], but `map_element` replaces the +/// element, so a node that rewrites what it produces still keeps every column +/// it never declares. +#[node_macro::node(category("Test"), extent(reverse_lanes_extent))] +fn scale_lanes<'e>( + ctx: impl Ctx + core_types::context::ExtractArena<'e> + ExtractIndex + InjectIndex + Copy, + content: IList, + opacity: f64, +) -> Result, Attr)>, Interrupt> { + let lane = ctx.innermost_index() as usize; + if lane >= content.len() { + return Err(GraphError::past_end().into()); + } + let scaled = content.element_ref(lane) * 2.; + Ok((content.lane(lane).map_element(scaled), Attr(opacity))) +} + #[node_macro::node(category("Test"))] fn source_opacity(_: impl Ctx, _: (), element: f64, opacity: f64) -> (f64, Attr) { (element, Attr(opacity)) @@ -531,6 +553,42 @@ mod tests { } } + /// Serves lanes carrying both a Transform no gather kernel declares and an + /// Opacity one does, so a carried column can be told apart from a written one. + struct LeveledCarriedSource { + layout: Layout, + rows: Vec<(f64, DAffine2, f64)>, + } + + impl Node for LeveledCarriedSource { + fn serve<'e, 'l>(&self, input: &C, slot: FrameClaim<'e, 'l>) -> GPoll> + where + C: ExtractArena, + { + let (element, transform, opacity) = self.rows[input.innermost_index() as usize % self.rows.len()]; + let mut frame = slot; + let arena = ExtractArena::arena(input); + if frame.element(element, arena).is_none() { + return GPoll::arena_exhausted(); + } + write_attr_at::(&mut frame, &self.layout, transform); + write_attr_at::(&mut frame, &self.layout, opacity); + // SAFETY: the writes above complete the record of this layout. + GPoll::Final(unsafe { frame.finish_served() }) + } + + fn extent_at<'x>(&self, _input: &C, _level: u8, _frames: &core_types::record::Frames<'x>) -> GPoll + where + C: ExtractArena, + { + GPoll::Final(Extent::Exactly(self.rows.len())) + } + + fn layout(&self) -> &Layout { + &self.layout + } + } + /// A leveled source that keeps its count to itself: the extent is a lower /// bound and lanes past the data answer the past-end signal. struct DrainSourceNode { @@ -1508,6 +1566,56 @@ mod tests { } } + #[test] + fn a_substituted_subject_keeps_the_lanes_other_columns() { + let arena = Arena::new(1 << 16).unwrap(); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let ctx = ContextImpl::root(&scope); + + // Each lane carries a Transform the kernel never declares and an Opacity + // it does, so the carried columns and the written one are distinguishable. + let layout = Layout::default().with_writes( + 1, + core_types::record::element_write::(), + &[core_types::record::FieldWrite::of::(0), core_types::record::FieldWrite::of::(0)], + ); + let frames = frames_for(&[&layout]); + let rows = [(1., 10., 0.1), (2., 30., 0.2), (3., 20., 0.3)]; + let content = LeveledCarriedSource { + layout: layout.clone(), + rows: rows + .iter() + .map(|&(element, x, opacity)| (element, DAffine2::from_translation(glam::DVec2::new(x, 0.)), opacity)) + .collect(), + }; + let node = install( + ScaleLanesNode::new(RecordSource::new(content, &layout, &layout), ValueSource::new(0.25)), + scale_lanes_layout_meta(), + &[Some(&layout)], + ); + + let out = Node::::layout(&node).clone(); + assert_eq!(out.depth, 1, "substituting the subject preserves the level's depth"); + assert!( + out.offset_of(::NAME, 0).is_some(), + "the undeclared attribute survives a substituted subject" + ); + + let head = ctx.index_head(); + for (lane, &(element, x, carried_opacity)) in rows.iter().enumerate() { + let GPoll::Final(served) = core_types::record::capture(&node, &ctx.promoted(&head, lane as u64), &frames) else { + panic!("expected a final record"); + }; + assert_eq!(served.element::(), element * 2., "lane {lane} takes the substituted element, not the source's"); + let transform: DAffine2 = served.attr::(); + assert_eq!(transform.translation.x, x, "lane {lane} still carries the undeclared transform"); + let opacity: f64 = served.attr::(); + assert_ne!(opacity, carried_opacity, "the declared write must not leave the carried opacity in place"); + assert_eq!(opacity, 0.25, "lane {lane} takes the declared write over the carried column"); + } + } + #[test] fn batch_lanes_match_per_lane_eval() { let arena = Arena::new(1 << 16).unwrap();