Claim the node frame once and close every exit through its drop

This commit is contained in:
Dennis Kobert
2026-08-28 19:11:38 +00:00
parent f4d37524c8
commit aef750153b
4 changed files with 242 additions and 163 deletions
+136 -42
View File
@@ -662,7 +662,7 @@ impl<'e, C, N: Node<C, Output = RecordValue<'e>>> RecordEdge<'e, C> for N {}
/// Builds an element-only record from a kernel's poll: inline layouts land /// Builds an element-only record from a kernel's poll: inline layouts land
/// in the value, larger ones spill to the record stack, arena exhaustion of /// in the value, larger ones spill to the record stack, arena exhaustion of
/// a parked element reports as an error poll. /// a parked element reports as an error poll.
pub fn lift_poll<'e, T: Send + Sync>(poll: GPoll<T>, layout: &Layout, arena: &'e crate::arena::Arena) -> GPoll<RecordValue<'e>> { pub(crate) fn lift_poll<'e, T: Send + Sync>(poll: GPoll<T>, layout: &Layout, arena: &'e crate::arena::Arena) -> GPoll<RecordValue<'e>> {
let build = |element: T| { let build = |element: T| {
if layout.frame_bytes() == 0 { if layout.frame_bytes() == 0 {
let mut value = RecordValue::zeroed(); let mut value = RecordValue::zeroed();
@@ -1075,7 +1075,7 @@ pub mod stack {
/// Releases everything above `frame`'s `bytes`-sized region, keeping the /// Releases everything above `frame`'s `bytes`-sized region, keeping the
/// region itself. A node reclaims its inputs' frames on return but leaves /// region itself. A node reclaims its inputs' frames on return but leaves
/// its own output readable for its consumer. /// its own output readable for its consumer.
pub fn truncate_above(frame: *mut u8, bytes: usize) { pub(crate) fn truncate_above(frame: *mut u8, bytes: usize) {
STACK.with(|stack| { STACK.with(|stack| {
let top = frame as usize - stack.base.get() as usize + bytes.next_multiple_of(8); let top = frame as usize - stack.base.get() as usize + bytes.next_multiple_of(8);
debug_assert!(top <= stack.sp.get(), "truncate target must lie within the claimed stack"); debug_assert!(top <= stack.sp.get(), "truncate target must lie within the claimed stack");
@@ -1117,15 +1117,19 @@ pub mod stack {
} }
} }
/// Releases everything claimed above `mark`; a pointer already at or
/// below it stays.
pub(crate) fn release_above(mark: usize) {
STACK.with(|stack| {
if mark < stack.sp.get() {
stack.sp.set(mark);
}
});
}
impl Drop for ScopeGuard { impl Drop for ScopeGuard {
fn drop(&mut self) { fn drop(&mut self) {
STACK.with(|stack| { release_above(self.mark);
// An interrupt close may already have rewound below the entry;
// the scope only ever releases, never re-claims.
if self.mark < stack.sp.get() {
stack.sp.set(self.mark);
}
});
} }
} }
@@ -1141,27 +1145,6 @@ pub mod stack {
} }
} }
/// Reclaims the frames an inline node's inputs push. An inline node returns
/// its output by value rather than on the stack, so it has no frame whose
/// `truncate_above` would release its inputs; this guard captures the entry
/// pointer and rewinds to it on drop instead. Inactive (a no-op) for spilled
/// nodes, which release their inputs through their own frame.
pub struct ReclaimGuard {
scope: Option<stack::ScopeGuard>,
}
impl ReclaimGuard {
/// # Safety
/// When `active`, the node must return its output by value, so that no
/// record into the region above the entry pointer is live once its eval
/// returns and the guard rewinds.
pub unsafe fn new(active: bool) -> Self {
Self {
scope: active.then(|| unsafe { stack::ScopeGuard::enter() }),
}
}
}
/// Serves one record of a layout: claims the frame, takes type-checked /// Serves one record of a layout: claims the frame, takes type-checked
/// element and attribute writes, and closes into the served [`RecordValue`]. /// element and attribute writes, and closes into the served [`RecordValue`].
/// Unwritten fields serve their census defaults. The single-record sibling /// Unwritten fields serve their census defaults. The single-record sibling
@@ -1258,12 +1241,9 @@ impl<'l, 'e> FrameBuilder<'l, 'e> {
pub fn finish(mut self) -> Option<RecordValue<'e>> { pub fn finish(mut self) -> Option<RecordValue<'e>> {
assert!(self.wrote_element || self.layout.element.size == 0, "the element serves before the frame closes"); assert!(self.wrote_element || self.layout.element.size == 0, "the element serves before the frame closes");
let value = match self.frame.take() { let value = match self.frame.take() {
Some(frame) => { // SAFETY: the frame was claimed for this layout and stays claimed,
stack::pop(frame); // keeping the frame contract for the consumer's release.
// SAFETY: the frame was claimed for this layout and stays Some(frame) => RecordValue::spilled(unsafe { Rec::new(frame.cast_const()) }),
// readable until the next claim.
RecordValue::spilled(unsafe { Rec::new(frame.cast_const()) })
}
None => std::mem::replace(&mut self.value, RecordValue::zeroed()), None => std::mem::replace(&mut self.value, RecordValue::zeroed()),
}; };
match self.exhausted { match self.exhausted {
@@ -1274,11 +1254,7 @@ impl<'l, 'e> FrameBuilder<'l, 'e> {
} }
impl Drop for FrameBuilder<'_, '_> { impl Drop for FrameBuilder<'_, '_> {
fn drop(&mut self) { fn drop(&mut self) {}
if let Some(frame) = self.frame {
stack::pop(frame);
}
}
} }
/// Field-by-field carry from `from`'s layout into `to`'s, computed at /// Field-by-field carry from `from`'s layout into `to`'s, computed at
@@ -1320,7 +1296,7 @@ pub unsafe fn write_field<T>(dst: *mut u8, offset: usize, value: T) {
/// `dst` must be the claimed frame (or inline scratch when `frame_bytes` is /// `dst` must be the claimed frame (or inline scratch when `frame_bytes` is
/// 0) of a record whose element is `T` and whose frame size is `frame_bytes`, /// 0) of a record whose element is `T` and whose frame size is `frame_bytes`,
/// with every carried field already written. /// with every carried field already written.
pub unsafe fn lift_poll_into<'e, T: Send + Sync>(poll: GPoll<T>, dst: *mut u8, frame_bytes: usize, arena: &'e crate::arena::Arena) -> GPoll<RecordValue<'e>> { pub(crate) unsafe fn lift_poll_into<'e, T: Send + Sync>(poll: GPoll<T>, dst: *mut u8, frame_bytes: usize, arena: &'e crate::arena::Arena) -> GPoll<RecordValue<'e>> {
let release = || { let release = || {
if frame_bytes != 0 { if frame_bytes != 0 {
stack::truncate_above(dst, frame_bytes); stack::truncate_above(dst, frame_bytes);
@@ -1715,6 +1691,124 @@ pub unsafe fn interrupt_frame(entry: usize, layout: &Layout) {
claim_frame(layout); claim_frame(layout);
} }
/// A node's own output frame, claimed at eval entry: the one closing surface
/// for every exit. Writes land through it, [`Self::lift`] and [`Self::finish`]
/// serve the record, and its drop releases everything claimed above the frame
/// while keeping the frame itself, so the frame contract holds on value,
/// error, and pending exits alike with no per-exit ritual.
pub struct FrameClaim<'l> {
layout: &'l Layout,
inline: RecordValue<'static>,
frame: Option<*mut u8>,
own_end: usize,
}
impl<'l> FrameClaim<'l> {
/// Claims the layout's frame at the stack pointer; an inline layout's
/// record builds in the value itself.
pub fn enter(layout: &'l Layout) -> Self {
let frame = match layout.frame_bytes() {
0 => None,
bytes => Some(stack::push(bytes)),
};
Self {
layout,
inline: RecordValue::zeroed(),
frame,
own_end: stack::sp(),
}
}
fn dst(&mut self) -> *mut u8 {
match self.frame {
Some(frame) => frame,
None => (&raw mut self.inline).cast(),
}
}
/// Asserts the served element matches the wired layout, so a node whose
/// layout never resolved (or resolved at another type) panics here
/// instead of writing past its frame.
fn check_element<T: dyn_any::StaticTypeSized>(&self) {
let (size, _) = element_dims::<T>();
assert!(
self.layout.element.size == size && self.layout.element.parked == element_parked::<T>() && self.layout.element.type_id == std::any::TypeId::of::<T::Static>(),
"the served element `{}` ({size} bytes) must match the wired layout ({} bytes)",
std::any::type_name::<T>(),
self.layout.element.size,
);
}
/// Carries the plan's fields from a source record into the frame.
///
/// # Safety
/// `src` must be a live record of the plan's source layout, and the plan
/// must be the wiring-resolved plan of this frame's layout.
pub unsafe fn carry(&mut self, src: Rec, plan: &[(usize, usize, usize)]) {
unsafe { apply_plan(src, self.dst(), plan) };
}
/// Writes a field at its wiring-resolved offset.
///
/// # Safety
/// `offset` must be this layout's resolved offset for a field of `T`.
pub unsafe fn attr_at<T>(&mut self, offset: usize, value: T) {
unsafe { write_field(self.dst(), offset, value) };
}
/// Writes the element; `None` reports arena exhaustion for a parked
/// element. Panics where the element does not match the wired layout.
pub fn element<T: Send + Sync + dyn_any::StaticTypeSized>(&mut self, value: T, arena: &crate::arena::Arena) -> Option<()> {
self.check_element::<T>();
// SAFETY: the frame is this layout's fresh claim and the element
// check pinned `T` to the layout's element slot.
unsafe { write_element(self.dst(), value, arena) }
}
/// Lifts a kernel's poll into the frame and closes it: the element
/// writes on value polls, every poll keeps the frame claimed, and arena
/// exhaustion of a parked element reports as an error poll. Panics where
/// the element does not match the wired layout.
pub fn lift<'e, T: Send + Sync + dyn_any::StaticTypeSized>(mut self, poll: GPoll<T>, arena: &'e crate::arena::Arena) -> GPoll<RecordValue<'e>> {
self.check_element::<T>();
let frame_bytes = self.layout.frame_bytes();
let dst = self.dst();
// SAFETY: the frame is this layout's fresh claim, the element check
// pinned `T`, and the drop keeps the frame contract on every poll.
unsafe { lift_poll_into(poll, dst, frame_bytes, arena) }
}
/// Copies a complete record of this layout into the frame, for serving
/// cached or published bytes.
///
/// # Safety
/// `src` must point at a live record of this layout whose parked
/// references outlive the serving evaluation.
pub unsafe fn fill_copy(&mut self, src: *const u8) {
unsafe { std::ptr::copy_nonoverlapping(src, self.dst(), self.layout.size) };
}
/// The served record. The frame stays claimed for the consumer; the drop
/// releases only what was claimed above it.
///
/// # Safety
/// The frame must hold a complete record of the layout, written through
/// the carry, element, and field writes.
pub unsafe fn finish<'e>(mut self) -> RecordValue<'e> {
match self.frame {
Some(frame) => RecordValue::spilled(unsafe { Rec::new(frame.cast_const()) }),
// SAFETY: the inline record is the value's own bytes.
None => unsafe { (&raw mut self.inline).cast::<RecordValue<'e>>().read() },
}
}
}
impl Drop for FrameClaim<'_> {
fn drop(&mut self) {
stack::release_above(self.own_end);
}
}
/// A record deep-copied out of its evaluation: the packed bytes plus owned /// A record deep-copied out of its evaluation: the packed bytes plus owned
/// clones of every parked payload, replayable into a later evaluation's /// clones of every parked payload, replayable into a later evaluation's
/// storage through the layout's erased glue. The layout stays with the /// storage through the layout's erased glue. The layout stays with the
+56 -103
View File
@@ -1224,12 +1224,29 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
.collect::<Vec<(usize, &AttributeRead)>>() .collect::<Vec<(usize, &AttributeRead)>>()
}; };
// Interrupt exits close the frame like every other exit: rewind to the // A serving node's interrupt exits close through the frame claim's drop;
// scope's entry pointer (claimed frames above it are dead once the exit // a forwarding node still closes by hand, since its kernel serves the
// carries no value) and claim the node's own frame. `_entry_sp` is bound // record and a top-level claim would double the frame.
// at the top of `eval` and per lane in the generated batch body. let claims_frame = flip || record_io;
let interrupt_close = quote! { let interrupt_close = match claims_frame {
unsafe { #core_types::record::interrupt_frame(_entry_sp, <Self as #core_types::node::Node<#ctx_ident>>::layout(self)) }; true => quote!(),
false => quote! {
unsafe { #core_types::record::interrupt_frame(_entry_sp, <Self as #core_types::node::Node<#ctx_ident>>::layout(self)) };
},
};
let frame_entry = match claims_frame {
true => quote! {
#[allow(unused_mut, unused_variables)]
let mut __frame = #core_types::record::FrameClaim::enter(<Self as #core_types::node::Node<#ctx_ident>>::layout(self));
},
false => quote!(let _entry_sp = #core_types::record::stack::sp();),
};
let lane_frame_entry = match claims_frame {
true => quote! {
#[allow(unused_mut, unused_variables)]
let mut __frame = #core_types::record::FrameClaim::enter(__node_layout);
},
false => quote!(let _entry_sp = #core_types::record::stack::sp();),
}; };
let bind_body = |index: usize, field: &ParsedField, batch_mode: bool| { let bind_body = |index: usize, field: &ParsedField, batch_mode: bool| {
let name = &field.pat_ident.ident; let name = &field.pat_ident.ident;
@@ -1737,17 +1754,12 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
}; };
let clamp = clamp_tokens(field); let clamp = clamp_tokens(field);
quote! { quote! {
let mut __carried = #core_types::record::RecordValue::zeroed();
let __dst = match self.__frame_bytes {
0 => __carried.as_mut_ptr(),
__bytes => #core_types::record::stack::push(__bytes),
};
let __src = match __cell.eval_input(0, &self.#name, __input) { let __src = match __cell.eval_input(0, &self.#name, __input) {
Ok(value) => value, Ok(value) => value,
Err(interrupt) => { #interrupt_close return interrupt.into(); } Err(interrupt) => return interrupt.into(),
}; };
let __src_rec = self.__in_0.rec(&__src); let __src_rec = self.__in_0.rec(&__src);
unsafe { #core_types::record::apply_plan(__src_rec, __dst, &self.__plan) }; unsafe { __frame.carry(__src_rec, &self.__plan) };
#read #read
#clamp #clamp
} }
@@ -1755,16 +1767,13 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
// Async slots persist plain values across evaluations; a flipped source // Async slots persist plain values across evaluations; a flipped source
// lifts the slot value onto its record wire at every merge point, into // lifts the slot value onto its record wire at every merge point, into
// the carried frame when the node has a carrier. // the carried frame when the node has a carrier.
let merge_lifted = |poll: TokenStream2| match (flip, carrier_flip) { let merge_lifted = |poll: TokenStream2| match flip {
(true, true) => { true => quote!(__cell.merge(__frame.lift(#poll, #core_types::context::ExtractArena::arena(__input)))),
quote!(__cell.merge(unsafe { #core_types::record::lift_poll_into(#poll, __dst, self.__frame_bytes, #core_types::context::ExtractArena::arena(__input)) })) false => quote!(__cell.merge(#poll)),
}
(true, false) => quote!(__cell.merge(#core_types::record::lift_poll(#poll, &self.__layout, #core_types::context::ExtractArena::arena(__input)))),
(false, _) => quote!(__cell.merge(#poll)),
}; };
let pending_return = match flip && carrier_flip { let pending_return = match flip && carrier_flip {
true => quote! { true => quote! {
unsafe { #core_types::record::lift_poll_into::<#slot_ty>(#core_types::gpoll::GPoll::Pending, __dst, self.__frame_bytes, #core_types::context::ExtractArena::arena(__input)) } __frame.lift::<#slot_ty>(#core_types::gpoll::GPoll::Pending, #core_types::context::ExtractArena::arena(__input))
}, },
false => quote!({ #interrupt_close #core_types::gpoll::GPoll::Pending }), false => quote!({ #interrupt_close #core_types::gpoll::GPoll::Pending }),
}; };
@@ -1854,13 +1863,13 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
let __src_rec = self.__carrier.rec(&__src); let __src_rec = self.__carrier.rec(&__src);
} }
}); });
let carry = (!skips_carrier && !lazy_carrier).then(|| quote!(unsafe { #core_types::record::apply_plan(__src_rec, __dst, &self.__plan) };)); let carry = (!skips_carrier && !lazy_carrier).then(|| quote!(unsafe { __frame.carry(__src_rec, &self.__plan) };));
// A lazy carrier's source record is the token the kernel returned; its // A lazy carrier's source record is the token the kernel returned; its
// content frames sit above `__dst` and stay readable until the truncate. // content frames sit above the claim and stay readable until its drop.
let lazy_carry = match lazy_carrier { let lazy_carry = match lazy_carrier {
true => quote! { true => quote! {
let __src_rec = self.__carrier.rec(&__element); let __src_rec = self.__carrier.rec(&__element);
unsafe { #core_types::record::apply_plan(__src_rec, __dst, &self.__plan) }; unsafe { __frame.carry(__src_rec, &self.__plan) };
}, },
false => TokenStream2::new(), false => TokenStream2::new(),
}; };
@@ -1868,7 +1877,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
let gather_carry = match gather_carrier { let gather_carry = match gather_carrier {
true => quote! { true => quote! {
let __src_rec = __element.rec(); let __src_rec = __element.rec();
unsafe { #core_types::record::apply_plan(__src_rec, __dst, &self.__plan) }; unsafe { __frame.carry(__src_rec, &self.__plan) };
}, },
false => TokenStream2::new(), false => TokenStream2::new(),
}; };
@@ -1918,7 +1927,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
let element_store = element_write.map(|ty| { let element_store = element_write.map(|ty| {
let ty = &crate::codegen::classify::substitute_lifetimes(ty, "'_"); let ty = &crate::codegen::classify::substitute_lifetimes(ty, "'_");
quote! { quote! {
if unsafe { #core_types::record::write_element::<#ty>(__dst, __element, #core_types::context::ExtractArena::arena(__input)) }.is_none() { if __frame.element::<#ty>(__element, #core_types::context::ExtractArena::arena(__input)).is_none() {
return #core_types::gpoll::Interrupt::from(#core_types::gpoll::GraphError { return #core_types::gpoll::Interrupt::from(#core_types::gpoll::GraphError {
kind: #core_types::gpoll::ErrorKind::ArenaExhausted, kind: #core_types::gpoll::ErrorKind::ArenaExhausted,
trace: ::std::vec::Vec::new(), trace: ::std::vec::Vec::new(),
@@ -1929,14 +1938,9 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
}); });
let attr_stores = attr_binders.iter().enumerate().map(|(index, binder)| { let attr_stores = attr_binders.iter().enumerate().map(|(index, binder)| {
let slot = format_ident!("__write_{index}"); let slot = format_ident!("__write_{index}");
quote!(unsafe { #core_types::record::write_field(__dst, self.#slot, #binder) };) quote!(unsafe { __frame.attr_at(self.#slot, #binder) };)
}); });
quote! { quote! {
let mut __value = #core_types::record::RecordValue::zeroed();
let __dst = match self.__frame_bytes {
0 => __value.as_mut_ptr(),
__bytes => #core_types::record::stack::push(__bytes),
};
#carrier_eval #carrier_eval
#carry #carry
#(#carrier_read_bindings)* #(#carrier_read_bindings)*
@@ -1946,10 +1950,8 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
#gather_carry #gather_carry
#element_store #element_store
#(#attr_stores)* #(#attr_stores)*
if self.__frame_bytes != 0 { // SAFETY: the carry and the writes above complete the record.
#core_types::record::stack::truncate_above(__dst, self.__frame_bytes); let __value = unsafe { __frame.finish() };
__value = #core_types::record::RecordValue::spilled(unsafe { #core_types::record::Rec::new(__dst.cast_const()) });
}
} }
}); });
let record_tail = record_tail_core.clone().map(|core| { let record_tail = record_tail_core.clone().map(|core| {
@@ -1960,49 +1962,26 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
}); });
let flip_tail = flip.then(|| { let flip_tail = flip.then(|| {
if matches!(*model, Dialect::Poll) { if matches!(*model, Dialect::Poll) {
return match &carried_prelude { let prelude = carried_prelude.clone().unwrap_or_default();
Some(prelude) => quote! { return quote! {
#prelude #prelude
__cell.merge(unsafe { #core_types::record::lift_poll_into(#kernel_call, __dst, self.__frame_bytes, #core_types::context::ExtractArena::arena(__input)) }) __cell.merge(__frame.lift(#kernel_call, #core_types::context::ExtractArena::arena(__input)))
},
None => quote! {
let mut __scratch = #core_types::record::RecordValue::zeroed();
let __dst = match self.__frame_bytes {
0 => __scratch.as_mut_ptr(),
__bytes => #core_types::record::stack::push(__bytes),
};
__cell.merge(unsafe { #core_types::record::lift_poll_into(#kernel_call, __dst, self.__frame_bytes, #core_types::context::ExtractArena::arena(__input)) })
},
}; };
} }
let kernel_value = match *model { let kernel_value = match *model {
Dialect::Interrupt => quote! { Dialect::Interrupt => quote! {
match #kernel_call { match #kernel_call {
Ok(value) => value, Ok(value) => value,
Err(interrupt) => { #interrupt_close return interrupt.into(); } Err(interrupt) => return interrupt.into(),
} }
}, },
_ => quote!(#kernel_call), _ => quote!(#kernel_call),
}; };
match &carried_prelude { let prelude = carried_prelude.clone().unwrap_or_default();
// The carrier evaluates beyond the claimed frame, so the kernel quote! {
// runs after the push. #prelude
Some(prelude) => quote! { let __kernel_value = #kernel_value;
#prelude __cell.merge(__frame.lift(#core_types::gpoll::GPoll::Final(__kernel_value), #core_types::context::ExtractArena::arena(__input)))
let __kernel_value = #kernel_value;
__cell.merge(unsafe {
#core_types::record::lift_poll_into(#core_types::gpoll::GPoll::Final(__kernel_value), __dst, self.__frame_bytes, #core_types::context::ExtractArena::arena(__input))
})
},
None => quote! {
let mut __scratch = #core_types::record::RecordValue::zeroed();
let __dst = match self.__frame_bytes {
0 => __scratch.as_mut_ptr(),
__bytes => #core_types::record::stack::push(__bytes),
};
let __kernel_value = #kernel_value;
__cell.merge(unsafe { #core_types::record::lift_poll_into(#core_types::gpoll::GPoll::Final(__kernel_value), __dst, self.__frame_bytes, #core_types::context::ExtractArena::arena(__input)) })
},
} }
}); });
let tail_form = if async_fn { let tail_form = if async_fn {
@@ -2117,8 +2096,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
true => { true => {
let mark = format_ident!("__scope_{index}"); let mark = format_ident!("__scope_{index}");
quote! { quote! {
// SAFETY: the bind copies its reads out by value; the // SAFETY: the bind's reads copy out by value.
// drop point matches the old rewind.
let #mark = unsafe { #core_types::record::stack::ScopeGuard::enter() }; let #mark = unsafe { #core_types::record::stack::ScopeGuard::enter() };
#body #body
drop(#mark); drop(#mark);
@@ -2198,10 +2176,10 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
for __lane in 0..__len { for __lane in 0..__len {
#core_types::context::InjectIndex::set_index(&mut __lane_ctx, __range.start + __lane as u64); #core_types::context::InjectIndex::set_index(&mut __lane_ctx, __range.start + __lane as u64);
let __input = &__lane_ctx; let __input = &__lane_ctx;
// SAFETY: the lane's record is copied out before the scope // SAFETY: the lane's record copies out before the scope
// releases it, and valueless exits serve nothing above it. // releases it.
let __lane_scope = unsafe { #core_types::record::stack::ScopeGuard::enter() }; let __lane_scope = unsafe { #core_types::record::stack::ScopeGuard::enter() };
let _entry_sp = #core_types::record::stack::sp(); #lane_frame_entry
let __cell = __cell.snapshot(); let __cell = __cell.snapshot();
#(#rebinds)* #(#rebinds)*
#(#binds)* #(#binds)*
@@ -2266,10 +2244,8 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
if __scratch.len() * 8 < __len * __stride { if __scratch.len() * 8 < __len * __stride {
return #core_types::node::BatchStatus::InvalidRange; return #core_types::node::BatchStatus::InvalidRange;
} }
// SAFETY: every lane copies into the caller's scratch, so // SAFETY: every lane copies into the caller's scratch.
// nothing served above the entry escapes the batch scope.
let __entry_scope = unsafe { #core_types::record::stack::ScopeGuard::enter() }; let __entry_scope = unsafe { #core_types::record::stack::ScopeGuard::enter() };
let _entry_sp = #core_types::record::stack::sp();
let __cell = #cell_constructor; let __cell = #cell_constructor;
let __base_ctx = { let __base_ctx = {
let mut __ctx = *__input; let mut __ctx = *__input;
@@ -2315,7 +2291,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
} }
// The element store parks droppable elements in the arena. // The element store parks droppable elements in the arena.
if let Some(ty) = element_write { if let Some(ty) = element_write {
bounds.push({ let ty = &crate::codegen::classify::substitute_lifetimes(ty, "'static"); quote!(#ty: ::core::marker::Send + ::core::marker::Sync + 'static) }); bounds.push({ let ty = &crate::codegen::classify::substitute_lifetimes(ty, "'static"); quote!(#ty: ::core::marker::Send + ::core::marker::Sync + #core_types::StaticTypeSized + 'static) });
} }
} }
// A routing node's value elements copy out of their records. // A routing node's value elements copy out of their records.
@@ -2357,7 +2333,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
}) })
.collect(); .collect();
let out = crate::codegen::classify::substitute_lifetimes(&slot_value_type(&parsed.output_type), "'static"); let out = crate::codegen::classify::substitute_lifetimes(&slot_value_type(&parsed.output_type), "'static");
bounds.push(quote!(#out: ::core::marker::Send + ::core::marker::Sync + 'static)); bounds.push(quote!(#out: ::core::marker::Send + ::core::marker::Sync + #core_types::StaticTypeSized + 'static));
bounds bounds
} }
false => Vec::new(), false => Vec::new(),
@@ -2571,27 +2547,6 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
} }
}); });
// An inline node only leaks if an input spilled a frame. Flip nodes store
// each input's layout, so gate precisely; inline record-io nodes are rare
// (attributes usually spill the output) and their value-input layouts are
// not all stored, so guard them whenever inline.
let reclaim_active = match flip {
true => {
let spilled_inputs = (0..regular_fields.len()).map(|index| {
let slot = format_ident!("__in_{index}");
quote!(self.#slot.frame_bytes() != 0)
});
quote!(self.__frame_bytes == 0 && (false #(|| #spilled_inputs)*))
}
false => quote!(self.__frame_bytes == 0),
};
let reclaim_guard = (flip || record_io).then(|| {
quote! {
// SAFETY: an inline node returns its output by value, so nothing above the entry pointer is live when the guard rewinds.
let __reclaim_guard = unsafe { #core_types::record::ReclaimGuard::new(#reclaim_active) };
}
});
// The eval body as an ordered step sequence: bind each input, clamp, then the // The eval body as an ordered step sequence: bind each input, clamp, then the
// tail. The record-stack mark/rewind of a read-out bind is applied here from // tail. The record-stack mark/rewind of a read-out bind is applied here from
// the role, so the discipline is structural rather than per-arm. // the role, so the discipline is structural rather than per-arm.
@@ -2617,8 +2572,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
true => { true => {
let mark = format_ident!("__scope_{index}"); let mark = format_ident!("__scope_{index}");
quote! { quote! {
// SAFETY: the bind copies its reads out by value; the drop // SAFETY: the bind's reads copy out by value.
// point matches the old rewind.
let #mark = unsafe { #core_types::record::stack::ScopeGuard::enter() }; let #mark = unsafe { #core_types::record::stack::ScopeGuard::enter() };
#body #body
drop(#mark); drop(#mark);
@@ -2665,9 +2619,8 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
false => None, false => None,
} }
}; };
let _entry_sp = #core_types::record::stack::sp(); #frame_entry
let __cell = #cell_constructor; let __cell = #cell_constructor;
#reclaim_guard
#(#eval_body)* #(#eval_body)*
} }
+10 -7
View File
@@ -2320,13 +2320,16 @@ mod tests {
Some(value) => (content_layout.clone(), vec![("opacity", value)]), Some(value) => (content_layout.clone(), vec![("opacity", value)]),
None => (f64_layout(&[]), vec![]), None => (f64_layout(&[]), vec![]),
}; };
let node = FallbackNode::new( let node = install_flip(
core_types::record::RecordLift::<(), _>::new(ValueNode(())), FallbackNode::new(
f64_record_source(&content_layout, 7., fields), core_types::record::RecordLift::<(), _>::new(ValueNode(())),
alternate, f64_record_source(&content_layout, 7., fields),
&unit_layout, alternate,
&content_layout, &unit_layout,
&alternate_layout, &content_layout,
&alternate_layout,
),
&f64_layout(&[]),
); );
let GPoll::Final(value) = node.eval(&ctx) else { let GPoll::Final(value) = node.eval(&ctx) else {
panic!("expected a final record"); panic!("expected a final record");
+40 -11
View File
@@ -1064,6 +1064,23 @@ mod graphene_test {
unsafe { layout.rec(value).element::<T>() } unsafe { layout.rec(value).element::<T>() }
} }
fn out_layout<T: Clone + Send + Sync + core_types::StaticTypeSized>() -> Layout
where
<T as core_types::StaticTypeSized>::Static: Clone + Send + Sync,
{
Layout::default().with_writes(0, core_types::record::element_write::<T>(), &[])
}
fn installed<N: Node<ContextImpl<'static>>>(mut node: N, layout: &Layout) -> N {
node.set_layout(core_types::record::RecordLayout {
frame_bytes: layout.frame_bytes(),
plan: Vec::new(),
layout: layout.clone(),
lane_invariant: u32::MAX,
});
node
}
#[test] #[test]
fn generated_add_evaluates_through_the_node_path() { fn generated_add_evaluates_through_the_node_path() {
let arena = Arena::new(64).unwrap(); let arena = Arena::new(64).unwrap();
@@ -1072,8 +1089,8 @@ mod graphene_test {
let (a, la) = lifted(SourceNode(1.0f64)); let (a, la) = lifted(SourceNode(1.0f64));
let (b, lb) = lifted(SourceNode(2.0f64)); let (b, lb) = lifted(SourceNode(2.0f64));
let graph = AddNode::<_, _, f64, f64>::new(a, b, &la, &lb); let out = out_layout::<f64>();
let out = Node::<ContextImpl>::layout(&graph).clone(); let graph = installed(AddNode::<_, _, f64, f64>::new(a, b, &la, &lb), &out);
reserve_for(&[&la, &lb, &out]); reserve_for(&[&la, &lb, &out]);
let GPoll::Final(value) = Node::eval(&graph, &ctx) else { let GPoll::Final(value) = Node::eval(&graph, &ctx) else {
@@ -1090,12 +1107,12 @@ mod graphene_test {
let (index, li) = lifted(IndexNode); let (index, li) = lifted(IndexNode);
let (src, ls) = lifted(SourceNode(10.0f64)); let (src, ls) = lifted(SourceNode(10.0f64));
let node = AddNode::<_, _, f64, f64>::new(index, src, &li, &ls); let out = out_layout::<f64>();
let out = Node::<ContextImpl>::layout(&node).clone(); let node = installed(AddNode::<_, _, f64, f64>::new(index, src, &li, &ls), &out);
reserve_for(&[&li, &ls, &out]); reserve_for(&[&li, &ls, &out]);
let erased: Box<ErasedRecordNode> = Box::new(node); let erased: Box<ErasedRecordNode> = Box::new(node);
// One u64 word per lane: the uninstalled layout keeps the f64 inline. // One u64 word per lane at the element-only layout.
let mut scratch = [const { MaybeUninit::uninit() }; 4]; let mut scratch = [const { MaybeUninit::uninit() }; 4];
let status = erased.eval_batch(&ctx, 2..6, Some(&mut scratch)); let status = erased.eval_batch(&ctx, 2..6, Some(&mut scratch));
let BatchStatus::Filled(batch, finality, _) = status else { let BatchStatus::Filled(batch, finality, _) = status else {
@@ -1114,8 +1131,14 @@ mod graphene_test {
let ctx = ContextImpl::root(&scope); let ctx = ContextImpl::root(&scope);
let entries = super::_logical_or_mod::logical_or_entries(); let entries = super::_logical_or_mod::logical_or_entries();
let wired = construct(&entries[0], vec![record_value_edge(true), record_value_edge(false)]).unwrap(); let mut wired = construct(&entries[0], vec![record_value_edge(true), record_value_edge(false)]).unwrap();
let layout = wired.layout().clone(); let layout = out_layout::<bool>();
wired.set_layout(core_types::record::RecordLayout {
frame_bytes: layout.frame_bytes(),
plan: Vec::new(),
layout: layout.clone(),
lane_invariant: u32::MAX,
});
let edge = wired.downcast_record::<bool>().unwrap(); let edge = wired.downcast_record::<bool>().unwrap();
reserve_for(&[&layout]); reserve_for(&[&layout]);
@@ -1154,8 +1177,14 @@ mod graphene_test {
); );
assert_eq!(entries[3].io.return_value, core_types::registry::record_type::<DVec2>()); assert_eq!(entries[3].io.return_value, core_types::registry::record_type::<DVec2>());
let wired = construct(&entries[0], vec![record_value_edge(1.5f64), record_value_edge(2.5f64)]).unwrap(); let mut wired = construct(&entries[0], vec![record_value_edge(1.5f64), record_value_edge(2.5f64)]).unwrap();
let layout = wired.layout().clone(); let layout = out_layout::<f64>();
wired.set_layout(core_types::record::RecordLayout {
frame_bytes: layout.frame_bytes(),
plan: Vec::new(),
layout: layout.clone(),
lane_invariant: u32::MAX,
});
let edge = wired.downcast_record::<f64>().unwrap(); let edge = wired.downcast_record::<f64>().unwrap();
reserve_for(&[&layout]); reserve_for(&[&layout]);
@@ -1306,8 +1335,8 @@ mod graphene_test {
let (fallback, lfb) = lifted(FallbackNode); let (fallback, lfb) = lifted(FallbackNode);
let (src, ls) = lifted(SourceNode(5.0f64)); let (src, ls) = lifted(SourceNode(5.0f64));
let graph = AddNode::<_, _, f64, f64>::new(fallback, src, &lfb, &ls); let out = out_layout::<f64>();
let out = Node::<ContextImpl>::layout(&graph).clone(); let graph = installed(AddNode::<_, _, f64, f64>::new(fallback, src, &lfb, &ls), &out);
reserve_for(&[&lfb, &ls, &out]); reserve_for(&[&lfb, &ls, &out]);
let GPoll::Fallback(boxed) = Node::eval(&graph, &ctx) else { let GPoll::Fallback(boxed) = Node::eval(&graph, &ctx) else {