From e415714b043528f0094f777b5708ba65c619b55a Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Tue, 1 Sep 2026 22:51:18 +0000 Subject: [PATCH] Promote a leaf element by moving its header into the persistent region --- node-graph/libraries/core-types/src/record.rs | 134 +++++++++++++++++- 1 file changed, 128 insertions(+), 6 deletions(-) diff --git a/node-graph/libraries/core-types/src/record.rs b/node-graph/libraries/core-types/src/record.rs index ef55cd924c..be6d1fd515 100644 --- a/node-graph/libraries/core-types/src/record.rs +++ b/node-graph/libraries/core-types/src/record.rs @@ -118,6 +118,10 @@ pub struct ElementWrite { pub type_id: std::any::TypeId, pub clone_out: unsafe fn(*const u8) -> Box, pub repark: unsafe fn(&(dyn std::any::Any + Send + Sync), *mut u8, &crate::arena::Arena) -> Option<()>, + /// Moves a parked payload's header into the persistent region rather than + /// cloning the heap it owns, returning the new header. `None` declines, + /// which leaves the caller its clone path. + pub park_move: unsafe fn(*const u8, &Promotion<'_>) -> Option<*const u8>, /// Hashes the element's content. `None` means the stored bytes are the /// content, which holds for every unparked element. pub content_hash: Option, @@ -141,6 +145,9 @@ impl Default for ElementWrite { unsafe fn repark(_value: &(dyn std::any::Any + Send + Sync), _dst: *mut u8, _arena: &crate::arena::Arena) -> Option<()> { Some(()) } + unsafe fn park_move(_parked: *const u8, _promotion: &Promotion<'_>) -> Option<*const u8> { + None + } Self { size: 0, align: 0, @@ -148,6 +155,7 @@ impl Default for ElementWrite { type_id: std::any::TypeId::of::<()>(), clone_out, repark, + park_move, content_hash: None, content_eq: None, } @@ -1453,6 +1461,23 @@ where let value = value.downcast_ref::().expect("an element replays at its own type"); unsafe { write_element_sized(dst, value.clone(), arena, retained) } } + /// Declines for a type carrying deep glue, whose value may hold interiors + /// the evaluation's arena owns and which a moved header may not reference. + /// + /// # Safety + /// `parked` must address a live payload of `T`. + unsafe fn park_move(parked: *const u8, promotion: &Promotion<'_>) -> Option<*const u8> + where + T::Static: Clone + Send + Sync, + { + deep_element_glue(std::any::TypeId::of::()).is_none().then_some(())?; + // SAFETY: the caller's contract, at the type the park was written with. + let value: &(dyn std::any::Any + Send + Sync) = unsafe { &*parked.cast::() }; + let retained = retained_measure(std::any::TypeId::of::()).map_or(0, |measure| measure(value)); + // SAFETY: as above, and the decline above establishes that the payload + // owns all of its content. + unsafe { promotion.move_park::(parked, retained) }.map(<*const T::Static>::cast) + } let (size, align) = element_dims::(); ElementWrite { size, @@ -1461,6 +1486,7 @@ where type_id: std::any::TypeId::of::(), clone_out: clone_out::, repark: repark::, + park_move: park_move::, content_hash: None, content_eq: None, } @@ -1823,6 +1849,26 @@ impl<'a> Promotion<'a> { pub fn evaluation_lived(&self, ptr: *const u8) -> bool { self.transient.contains(ptr) || (ptr as usize).wrapping_sub(self.frames.0) < self.frames.1 } + + /// Moves a transient payload's header into the persistent region instead of + /// cloning the heap it owns: the heap travels with the drop obligation and + /// is freed at the persistent flush, never at the transient reset. `None` + /// where the header is not the transient arena's own park or the region + /// refused it, which leaves the caller its clone path. + /// + /// The forwarding is the evaluation's, so a payload two records share moves + /// once and both reach the one persistent header. The source header stays + /// readable to the evaluation's remaining sharers, which are the only reads + /// the move keeps sound: no read of it may outlive the persistent flush. + /// + /// # Safety + /// `parked` must address a live `T` the transient arena parked, and `T` + /// must own all of its content, a persistent header being allowed to + /// reference no transient storage. + pub unsafe fn move_park(&self, parked: *const u8, retained: usize) -> Option<*const T> { + // SAFETY: the caller's contract, forwarded to the parking arena. + unsafe { self.transient.move_park::(parked, self.persistent, retained) } + } } /// Rewrites one promoted record's parked references in place: the lane bytes @@ -1845,11 +1891,16 @@ unsafe fn promote_record(layout: &Layout, dst: *mut u8, promotion: &Promotion<'_ match element_promote_glue(layout.element.type_id) { // SAFETY: the slot images a parked element of this type. Some(promote) => unsafe { promote(dst.cast_const(), dst, promotion) }?, - None => { - // SAFETY: as above, and the clone owns its content. - let owned = unsafe { (layout.element.clone_out)(dst.cast_const()) }; - unsafe { (layout.element.repark)(&*owned, dst, promotion.persistent) }?; - } + // SAFETY: as above; the header is the payload's own. + None => match unsafe { (layout.element.park_move)(parked, promotion) } { + // SAFETY: a parked element slot holds one reference at offset 0. + Some(moved) => unsafe { dst.cast::<*const u8>().write(moved) }, + None => { + // SAFETY: as above, and the clone owns its content. + let owned = unsafe { (layout.element.clone_out)(dst.cast_const()) }; + unsafe { (layout.element.repark)(&*owned, dst, promotion.persistent) }?; + } + }, } } } @@ -3082,10 +3133,81 @@ mod tests { let published = span.batch(&persistent, &layout).unwrap(); let promoted = unsafe { published.get(0).rec().ptr().cast::<*const u8>().read() }; assert!(persistent.contains(promoted), "the promote re-parked the payload into the persistent region"); - assert_ne!(source, promoted, "an evaluation-lived payload is cloned, not shared"); + assert_ne!(source, promoted, "an evaluation-lived payload gets its own persistent header"); assert_eq!(unsafe { borrow_element::(published.get(0).rec()) }, "parked in the evaluation"); } + /// The measure production registers for `String`, without which a promote + /// credits the region 0 and the counters cannot be observed to transfer. + fn measure_strings() { + register_retained_heap::(|value| value.downcast_ref::().map_or(0, String::len)); + } + + #[test] + fn a_promoted_payload_moves_its_heap_rather_than_cloning_it() { + measure_strings(); + let mut transient = crate::arena::Arena::new(4096).unwrap(); + let persistent = crate::arena::Arena::new(4096).unwrap(); + let layout = Layout::default().with_writes(0, element_write::(), &[]); + + let mut buffer = vec![0u64; layout.frame_bytes().div_ceil(8)]; + let base = buffer.as_mut_ptr().cast::(); + let bounds = (base as usize, buffer.len() * 8); + let owned = String::from("the obligation travels with the header"); + let (heap, length) = (owned.as_ptr(), owned.len()); + unsafe { write_element_sized(base, owned, &transient, length) }.unwrap(); + assert_eq!(transient.retained_heap(), length); + + let batch = unsafe { crate::node::RecordBatch::new(base.cast_const(), 1, &layout) }; + let promotion = Promotion::new(&transient, bounds, &persistent); + let span = unsafe { MaterializedSpan::to_persistent(&batch, &promotion) }.unwrap(); + + let published = span.batch(&persistent, &layout).unwrap(); + let served = unsafe { borrow_element::(published.get(0).rec()) }; + assert_eq!(served.as_ptr(), heap, "the promote moved the header, so the served view names the pre-promote heap"); + assert_eq!(transient.retained_heap(), 0, "the transient counter gave the hint up"); + assert_eq!(persistent.retained_heap(), length, "and the persistent counter took it"); + + transient.reset(); + let published = span.batch(&persistent, &layout).unwrap(); + assert_eq!( + unsafe { borrow_element::(published.get(0).rec()) }, + "the obligation travels with the header", + "the moved payload survives the transient reset" + ); + } + + #[test] + fn a_payload_two_records_share_promotes_to_one_persistent_header() { + measure_strings(); + let transient = crate::arena::Arena::new(4096).unwrap(); + let mut persistent = crate::arena::Arena::new(4096).unwrap(); + let layout = Layout::default().with_writes(0, element_write::(), &[]); + let stride = layout.lane_stride(); + + let mut buffer = vec![0u64; (2 * stride).div_ceil(8)]; + let base = buffer.as_mut_ptr().cast::(); + let bounds = (base as usize, buffer.len() * 8); + let length = "shared across lanes".len(); + unsafe { write_element_sized(base, String::from("shared across lanes"), &transient, length) }.unwrap(); + // A carried field byte-copies the reference, so both lanes name the one park. + let shared = unsafe { base.cast::<*const u8>().read() }; + unsafe { base.add(stride).cast::<*const u8>().write(shared) }; + + let batch = unsafe { crate::node::RecordBatch::new(base.cast_const(), 2, &layout) }; + let promotion = Promotion::new(&transient, bounds, &persistent); + let span = unsafe { MaterializedSpan::to_persistent(&batch, &promotion) }.unwrap(); + + let published = span.batch(&persistent, &layout).unwrap(); + let first = unsafe { published.get(0).rec().ptr().cast::<*const u8>().read() }; + let second = unsafe { published.get(1).rec().ptr().cast::<*const u8>().read() }; + assert_eq!(first, second, "a payload two records share moves once"); + assert_eq!(persistent.retained_heap(), length, "and its hint transfers once"); + + persistent.reset(); + assert_eq!(persistent.retained_heap(), 0, "the flush frees the one header exactly once"); + } + #[test] fn a_persistent_element_is_shared_by_the_promote() { let transient = crate::arena::Arena::new(4096).unwrap();