From 29052a58db931ae2af70e7b8bea995fbff6b369a Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Sat, 29 Aug 2026 23:00:17 +0000 Subject: [PATCH] Let async source kernels write attributes through their frame claim --- node-graph/node-macro/src/codegen.rs | 54 +++++++++++-- node-graph/node-macro/src/codegen/classify.rs | 40 ++++++++-- node-graph/node-macro/src/codegen/ir.rs | 43 ++++++++++ node-graph/node-macro/src/validation.rs | 8 +- node-graph/nodes/gcore/src/record.rs | 79 +++++++++++++++++++ 5 files changed, 209 insertions(+), 15 deletions(-) diff --git a/node-graph/node-macro/src/codegen.rs b/node-graph/node-macro/src/codegen.rs index ab0e506986..ed01a7f548 100644 --- a/node-graph/node-macro/src/codegen.rs +++ b/node-graph/node-macro/src/codegen.rs @@ -291,7 +291,7 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn })); let async_source = parsed.injects_async_source_fields(); - let slot_value_type = crate::codegen::classify::substitute_lifetimes(&slot_value_type(output_type), "'static"); + let slot_value_type = crate::codegen::classify::substitute_lifetimes(&crate::codegen::classify::slot_static_type(output_type), "'static"); let slot_field = async_source .then(|| quote! { pub(super) slot: std::sync::Arc>>>> }) .into_iter(); @@ -1206,7 +1206,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn // The slot persists the plain value even on record wires, so the Clone // bound targets the slot type, not the (possibly lifted) trait output. - let slot_ty = slot_value_type(&parsed.output_type); + let slot_ty = crate::codegen::classify::slot_static_type(&parsed.output_type); let mut async_bounds = match (async_fn, future_kernel) { (false, false) => Vec::new(), (false, true) => vec![quote!(#slot_ty: Clone)], @@ -1721,9 +1721,10 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn let kernel_fields: Vec<&&ParsedField> = regular_fields.iter().filter(|field| !injected_name(&field.pat_ident.ident)).collect(); // A bare `Attr` in the return type cannot elide its lifetime, so the // kernel gets a fresh one; reference-valued writes name their real - // lifetime explicitly and pass through untouched. - let attr_injected = record_io.then(|| inject_attr_lifetimes(&parsed.output_type)).flatten(); - let attr_lifetime = attr_injected.is_some().then(|| quote!('__attr,)); + // lifetime explicitly and pass through untouched. An async source's value + // outlives the evaluation, so its writes are `'static` instead. + let attr_injected = record_io.then(|| inject_attr_lifetimes(&parsed.output_type, if async_source { "'static" } else { "'__attr" })).flatten(); + let attr_lifetime = (attr_injected.is_some() && !async_source).then(|| quote!('__attr,)); let lane_injected = gather_carrier .then(|| crate::codegen::classify::inject_lane_lifetime(attr_injected.as_ref().unwrap_or(&parsed.output_type))) .flatten(); @@ -1767,7 +1768,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn let params = snapshot_param.chain(data_kernel_params).chain(value_kernel_params); quote! { #[allow(clippy::too_many_arguments, clippy::type_complexity)] - #vis async fn #fn_name<#(#kernel_generics,)*>(#(#params),*) -> #output_type #fn_where #body + #vis async fn #fn_name<#(#kernel_generics,)*>(#(#params),*) -> #kernel_output #fn_where #body } } }; @@ -1835,10 +1836,47 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn #clamp } }); + // A writing source's carrier is its record-io carrier, so the fields it + // passes through ride the record plan rather than the flip one. + let carried_prelude = carried_prelude.or_else(|| { + (record_io && async_source && !skips_carrier).then(|| { + let field = regular_fields[0]; + let name = &field.pat_ident.ident; + let ty = carrier_read_ty.expect("a carrying record source reads a concrete element"); + quote! { + let __src = match __cell.eval_input(0, &self.#name, __input, __frame.frames()) { + Ok(value) => value, + Err(interrupt) => return interrupt.into(), + }; + let __src_rec = self.__carrier.rec(&__src); + unsafe { __frame.carry(__src_rec, &self.__plan) }; + let #name: #ty = unsafe { #core_types::record::read_element(__src_rec) }; + } + }) + }); // Async slots persist plain values across evaluations; the source lifts // the slot value onto its record wire at every merge point, into the // carried frame when the node has a carrier. - let merge_lifted = |poll: TokenStream2| quote!(__cell.merge(__frame.lift_served(#poll, #core_types::context::ExtractArena::arena(__input)))); + // A writing source stores the kernel's whole tuple as that plain value: + // the lift writes the attributes through the claim, then lifts the + // element, the shape the sync record tail closes with. + let source_writes = (record_io && async_source && !write_markers.is_empty()).then(|| { + let binders: Vec = (0..write_markers.len()).map(|index| format_ident!("__attr_{index}")).collect(); + let slots: Vec = (0..write_markers.len()).map(|index| format_ident!("__write_{index}")).collect(); + quote! { + .map(|(__element #(, #core_types::attribute::Attr(#binders))*)| { + #(unsafe { __frame.attr_at(self.#slots, #binders) };)* + __element + }) + } + }); + let merge_lifted = |poll: TokenStream2| match &source_writes { + None => quote!(__cell.merge(__frame.lift_served(#poll, #core_types::context::ExtractArena::arena(__input)))), + Some(writes) => quote! {{ + let __lifted = (#poll) #writes; + __cell.merge(__frame.lift_served(__lifted, #core_types::context::ExtractArena::arena(__input))) + }}, + }; // The claim drops with the frame still claimed, so a valueless exit needs // no closing of its own. let pending_return = quote!(#core_types::gpoll::GPoll::Pending); @@ -2568,6 +2606,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn let slot = format_ident!("__mat_cache_{index}"); quote!(#slot: ::core::default::Default::default(),) }); + let slot_default = async_source.then(|| quote!(slot: ::core::default::Default::default(),)).into_iter(); // A ranked input's element generic rides the struct as a phantom // parameter, so the constructor declares and initializes it too. let carried_type_params: Vec<&Ident> = struct_type_params @@ -2610,6 +2649,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn #(#read_names)* #(#write_defaults)* #(#mat_cache_defaults)* + #(#slot_default)* } } } diff --git a/node-graph/node-macro/src/codegen/classify.rs b/node-graph/node-macro/src/codegen/classify.rs index 3d77630a7a..afe534356f 100644 --- a/node-graph/node-macro/src/codegen/classify.rs +++ b/node-graph/node-macro/src/codegen/classify.rs @@ -206,18 +206,19 @@ pub(crate) fn substitute_routing_record(output: &Type, generic: &Ident, core_typ ty } -pub(crate) fn inject_attr_lifetimes(output: &Type) -> Option { - struct Injector { +pub(crate) fn inject_attr_lifetimes(output: &Type, lifetime: &str) -> Option { + struct Injector<'a> { changed: bool, + lifetime: &'a str, } - impl VisitMut for Injector { + impl VisitMut for Injector<'_> { fn visit_path_segment_mut(&mut self, segment: &mut syn::PathSegment) { if segment.ident == "Attr" && let PathArguments::AngleBracketed(args) = &mut segment.arguments && !args.args.iter().any(|arg| matches!(arg, GenericArgument::Lifetime(_))) { - args.args.insert(0, GenericArgument::Lifetime(Lifetime::new("'__attr", proc_macro2::Span::call_site()))); + args.args.insert(0, GenericArgument::Lifetime(Lifetime::new(self.lifetime, proc_macro2::Span::call_site()))); self.changed = true; } syn::visit_mut::visit_path_segment_mut(self, segment); @@ -225,7 +226,7 @@ pub(crate) fn inject_attr_lifetimes(output: &Type) -> Option { } let mut ty = output.clone(); - let mut injector = Injector { changed: false }; + let mut injector = Injector { changed: false, lifetime }; injector.visit_type_mut(&mut ty); injector.changed.then_some(ty) } @@ -281,9 +282,11 @@ pub(crate) fn unbounded_generic(parsed: &ParsedNodeFn, ty: &Type) -> Option Option { + let source = is_async_source(parsed); let value = match kernel_kind(&parsed.output_type) { KernelKind::Plain => parsed.output_type.clone(), KernelKind::Interrupt(inner) => inner, + _ if source => slot_value_type(&parsed.output_type), _ => return None, }; let writes = record_writes(&value); @@ -291,7 +294,9 @@ pub(crate) fn record_shape(parsed: &ParsedNodeFn) -> Option { if !has_reads && writes.is_none() { return None; } - if parsed.is_async { + // An async source's slot stores the kernel's plain tuple; the per-eval lift + // writes it through the claim, and the reads have no wire to bind against. + if source && (has_reads || writes.is_none()) { return None; } let carrier_field = parsed.fields.first()?; @@ -371,6 +376,11 @@ pub(crate) fn record_shape(parsed: &ParsedNodeFn) -> Option { if matches!(carrier, RecordCarrier::None) && !removes.is_empty() { return None; } + // The byte-carried token never becomes a value, so it cannot cross a + // future boundary. + if source && matches!(carrier, RecordCarrier::Token) { + return None; + } Some(RecordShape { carrier }) } @@ -605,6 +615,24 @@ pub(crate) fn is_source_kernel(output: &Type) -> bool { matches!(kernel_kind(output), KernelKind::Future(_) | KernelKind::FutureInterrupt(_)) } +/// A kernel whose value completes off the evaluation: an `async fn` or a +/// `SourceFuture` return. Its slot persists a plain value across evaluations, +/// so nothing it returns may borrow the arena. +pub(crate) fn is_async_source(parsed: &ParsedNodeFn) -> bool { + parsed.is_async || is_source_kernel(&parsed.output_type) +} + +/// The type an async source's slot persists. A writing source's value outlives +/// the evaluation, so every lifetime it names, including a bare `Attr`'s +/// elided one, is `'static`. +pub(crate) fn slot_static_type(output: &Type) -> Type { + let value = slot_value_type(output); + match record_writes(&value).is_some() { + true => substitute_lifetimes(&inject_attr_lifetimes(&value, "'static").unwrap_or_else(|| value.clone()), "'static"), + false => value, + } +} + pub(crate) enum KernelKind { Plain, Interrupt(Type), diff --git a/node-graph/node-macro/src/codegen/ir.rs b/node-graph/node-macro/src/codegen/ir.rs index 22f477e4fc..643d291dee 100644 --- a/node-graph/node-macro/src/codegen/ir.rs +++ b/node-graph/node-macro/src/codegen/ir.rs @@ -754,6 +754,49 @@ mod tests { ); } + #[test] + fn bridge_record_write_async_source() { + let node = assert_bridge( + quote!(category("")), + quote!( + async fn set_opacity_async(_: impl Ctx, val: f64) -> (f64, Attr) { + (val, Attr(1.)) + } + ), + ); + assert!(matches!(node_kind(&node), NodeKind::RecordIo), "a writing async source takes the record tail"); + assert!(matches!(node.effect, Effect::AsyncSource), "the writes do not change the effect axis"); + } + + #[test] + fn bridge_record_fresh_async_source() { + assert_bridge( + quote!(category("")), + quote!( + async fn make_async(_: impl Ctx, _: (), fill: f64) -> (f64, Attr) { + (fill, Attr(1.)) + } + ), + ); + } + + /// An async source's element is the value its slot stores, so a byte-carried + /// generic token has no form here. + #[test] + fn a_generic_token_carrier_has_no_async_source_form() { + let mut parsed = parse_node_fn( + quote!(category("")), + quote!( + async fn tag(_: impl Ctx, val: T) -> (T, Attr) { + (val, Attr(1.)) + } + ), + ) + .unwrap(); + parsed.replace_impl_trait_in_input(); + assert!(record_shape(&parsed).is_none()); + } + #[test] fn bridge_routing() { assert_bridge( diff --git a/node-graph/node-macro/src/validation.rs b/node-graph/node-macro/src/validation.rs index a2384a9bac..9bf20f0e8b 100644 --- a/node-graph/node-macro/src/validation.rs +++ b/node-graph/node-macro/src/validation.rs @@ -44,8 +44,9 @@ fn validate_record_io(parsed: &ParsedNodeFn) { return; } - if parsed.is_async || crate::codegen::is_source_kernel(&parsed.output_type) { - emit_error!(parsed.output_type.span(), "attribute io is not supported on async source kernels"); + let async_source = crate::codegen::classify::is_async_source(parsed); + if async_source && has_reads { + emit_error!(parsed.fn_name.span(), "attribute reads are not supported on async source kernels, only writes"); } if crate::codegen::is_poll_kernel(&parsed.output_type) { emit_error!(parsed.output_type.span(), "attribute io needs a plain or `Result<_, Interrupt>` kernel, not a `GPoll` one"); @@ -125,6 +126,9 @@ fn validate_record_io(parsed: &ParsedNodeFn) { (false, ParsedFieldType::Node(NodeParsedField { output_type, .. })) if lazy_carrier => crate::codegen::unbounded_generic(parsed, output_type), _ => None, }; + if async_source && token.is_some() { + emit_error!(carrier.pat_ident.span(), "an async source's element crosses the future boundary as a value; a passthrough generic element has none"); + } let element = writes.as_ref().map(|writes| &writes.element).unwrap_or(&value); match &token { Some(token) => { diff --git a/node-graph/nodes/gcore/src/record.rs b/node-graph/nodes/gcore/src/record.rs index 15e410bb20..3377005ec4 100644 --- a/node-graph/nodes/gcore/src/record.rs +++ b/node-graph/nodes/gcore/src/record.rs @@ -366,6 +366,19 @@ async fn double_async(_: impl Ctx, element: f64) -> f64 { element * 2. } +/// Test-only writing async source over a carrier: the slot stores the kernel's +/// whole tuple and the per-eval lift writes it through the claim. +#[node_macro::node(category("Test"))] +async fn fade_async(_: impl Ctx, element: f64, opacity: f64) -> (f64, Attr) { + (element * 2., Attr(opacity)) +} + +/// Test-only writing async source without a carrier: a fresh record per eval. +#[node_macro::node(category("Test"))] +async fn measure_async(_: impl Ctx, _: (), element: f64) -> (f64, Attr) { + (element, Attr(element.abs())) +} + #[node_macro::node(category("Test"))] fn fallback(ctx: impl Ctx, _: (), #[expose] content: impl Node, Output = (f64, Attr)>, #[expose] alternate: impl Node, Output = f64>) -> Result { let (element, opacity) = content.eval(ctx)?; @@ -2373,6 +2386,72 @@ mod tests { assert_eq!(served.attr::(), 0.25, "the fields re-carry on every eval"); } + #[test] + fn a_writing_async_source_lifts_its_slot_tuple_through_the_claim() { + let arena = Arena::new(1024).unwrap(); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let ctx = ContextImpl::root(&scope); + + let source_layout = f64_layout(&["length"]); + let (runtime, _) = lifted_value(core_types::runtime::RuntimeHandle(std::sync::Arc::new(InlineRuntime))); + let (source_id, _) = lifted_value(7 as SourceId); + let frames = frames_for(&[&source_layout]); + + let node = install( + FadeAsyncNode::new( + f64_record_source(&source_layout, 3., vec![("length", 9.)]), + ValueSource::new(0.5), + runtime, + source_id, + &source_layout, + ), + fade_async_layout_meta(), + &[Some(&source_layout)], + ); + assert_eq!(Node::::layout(&node), &fade_async_layout(&source_layout)); + + let GPoll::Final(served) = core_types::record::capture(&node, &ctx, &frames) else { + panic!("an inline completion is final on the spawning eval"); + }; + assert_eq!(served.element::(), 6.); + assert_eq!(served.attr::(), 0.5, "the write lands through the claim, not the future"); + assert_eq!(served.attr::(), 9., "the carrier's other fields still pass through"); + + let GPoll::Final(served) = core_types::record::capture(&node, &ctx, &frames) else { + panic!("a slot hit is final"); + }; + assert_eq!(served.element::(), 6., "the slot hit replays the element"); + assert_eq!(served.attr::(), 0.5, "the slot hit replays the write"); + assert_eq!(served.attr::(), 9.); + } + + #[test] + fn a_writing_async_source_without_a_carrier_writes_a_fresh_record() { + let arena = Arena::new(1024).unwrap(); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let ctx = ContextImpl::root(&scope); + + let (runtime, _) = lifted_value(core_types::runtime::RuntimeHandle(std::sync::Arc::new(InlineRuntime))); + let (source_id, _) = lifted_value(7 as SourceId); + let layout = measure_async_layout(); + let frames = frames_for(&[&layout]); + + let node = install( + MeasureAsyncNode::new(ValueSource::new(()), ValueSource::new(-4.), runtime, source_id), + measure_async_layout_meta(), + &[], + ); + assert_eq!(Node::::layout(&node), &layout); + + let GPoll::Final(served) = core_types::record::capture(&node, &ctx, &frames) else { + panic!("an inline completion is final on the spawning eval"); + }; + assert_eq!(served.element::(), -4.); + assert_eq!(served.attr::(), 4.); + } + #[test] fn lazy_reads_bind_to_their_edge_and_leave_the_untaken_branch_unevaluated() { let arena = Arena::new(1024).unwrap();