diff --git a/node-graph/libraries/core-types/src/record.rs b/node-graph/libraries/core-types/src/record.rs index d39253854e..857b9124ad 100644 --- a/node-graph/libraries/core-types/src/record.rs +++ b/node-graph/libraries/core-types/src/record.rs @@ -1465,23 +1465,47 @@ pub unsafe fn copy_record_bytes(layout: &Layout, rec: Rec) -> Box<[u8]> { unsafe { std::slice::from_raw_parts(rec.ptr(), layout.size) }.into() } -/// Builds a record value over `bytes`, a record of `layout` copied out -/// earlier: inline layouts copy into the value, spilled ones alias the -/// bytes. +/// Serves a record whose frame lives in storage the current evaluation does +/// not own (a cached batch, published bytes): claims the node's frame over a +/// copy of the source frame, so the contract that every node advances the +/// record stack by exactly its own frame holds for cache hits too. /// /// # Safety -/// `bytes` must hold a record of `layout` whose parked references are still -/// live; both hold for a copy taken in the same evaluation frame. -pub unsafe fn record_from_bytes<'e>(layout: &Layout, bytes: &'e [u8]) -> RecordValue<'e> { +/// `src` must point at a live record of `layout` whose parked references +/// outlive the serving evaluation. +pub unsafe fn serve_frame<'e>(layout: &Layout, src: *const u8) -> RecordValue<'e> { if layout.frame_bytes() == 0 { let mut value = RecordValue::zeroed(); - unsafe { std::ptr::copy_nonoverlapping(bytes.as_ptr(), value.as_mut_ptr(), bytes.len()) }; + unsafe { std::ptr::copy_nonoverlapping(src, value.as_mut_ptr(), layout.size) }; value } else { - RecordValue::spilled(unsafe { Rec::new(bytes.as_ptr()) }) + let dst = stack::push(layout.frame_bytes()); + unsafe { std::ptr::copy_nonoverlapping(src, dst, layout.size) }; + RecordValue::spilled(unsafe { Rec::new(dst.cast_const()) }) } } +/// Claims a node's frame with nothing to serve, for exits that yield no +/// record (past-end, pending): the frame contract holds on every exit, +/// error included. +pub fn claim_frame(layout: &Layout) { + if layout.frame_bytes() != 0 { + stack::push(layout.frame_bytes()); + } +} + +/// Closes an interrupted eval: rewinds to the eval's entry pointer (interrupt +/// exits carry no value, so frames claimed above it are dead) and claims the +/// node's own frame, keeping the frame contract on interrupt exits. +/// +/// # Safety +/// `entry` must be the stack pointer at the eval's entry, and no record +/// above it may be referenced after the close. +pub unsafe fn interrupt_frame(entry: usize, layout: &Layout) { + unsafe { stack::rewind(entry) }; + claim_frame(layout); +} + /// A captured record: the layout plus a generation-checked handle to the /// arena copy, materialized by the introspection holder, which owns the /// arena. A dead generation materializes to `None`, never to a stale read. @@ -1700,7 +1724,12 @@ where type Output = El; fn eval(&self, input: &C) -> GPoll { - self.edge.eval(input).map(|value| unsafe { read_element::(self.layout.rec(&value)) }) + let mark = stack::sp(); + let result = self.edge.eval(input).map(|value| unsafe { read_element::(self.layout.rec(&value)) }); + // SAFETY: the element copied out by value, so no record above the mark + // (the edge's frame) is live; a plain output claims no frame itself. + unsafe { stack::rewind(mark) }; + result } } @@ -1713,15 +1742,26 @@ where fn eval(&self, input: &C) -> GPoll> { match &self.plan { None => self.edge.eval(input), - Some(plan) if plan.union.frame_bytes() == 0 => self.edge.eval(input).map(|value| { - let mut out = RecordValue::zeroed(); - unsafe { plan.translate(plan.source.rec(&value), out.as_mut_ptr()) }; - out - }), + Some(plan) if plan.union.frame_bytes() == 0 => { + let mark = stack::sp(); + let result = self.edge.eval(input).map(|value| { + let mut out = RecordValue::zeroed(); + unsafe { plan.translate(plan.source.rec(&value), out.as_mut_ptr()) }; + out + }); + // SAFETY: the translation copied the record into the inline + // value, so no record above the mark is live. + unsafe { stack::rewind(mark) }; + result + } Some(plan) => { let dst = stack::push(plan.union.frame_bytes()); let value = self.edge.eval(input); - value.map(|value| RecordValue::spilled(unsafe { plan.translate(plan.source.rec(&value), dst) })) + let result = value.map(|value| RecordValue::spilled(unsafe { plan.translate(plan.source.rec(&value), dst) })); + // The source's frame dies with the translation; the claimed + // frame above stays as this node's output. + stack::truncate_above(dst, plan.union.frame_bytes()); + result } } } diff --git a/node-graph/nodes/gcore/src/memo.rs b/node-graph/nodes/gcore/src/memo.rs index 45dbef1d53..d4120ce611 100644 --- a/node-graph/nodes/gcore/src/memo.rs +++ b/node-graph/nodes/gcore/src/memo.rs @@ -4,7 +4,7 @@ use core_types::frame_table::{FrameTable, Lookup}; use core_types::gpoll::{Finality, GPoll}; use core_types::graphene_hash::CacheHash; use core_types::memo::IORecord; -use core_types::record::{LevelStatus, OwnedRecord, RecordCapture, RecordValue, copy_record_bytes, record_from_bytes}; +use core_types::record::{LevelStatus, OwnedRecord, RecordCapture, RecordValue, claim_frame, copy_record_bytes, serve_frame}; use core_types::registry::cache_key; use std::sync::Arc; use std::sync::Mutex; @@ -58,13 +58,15 @@ fn memoize<'e>( let serve = |entry: &MemoLevel| { if lane >= entry.lanes.len() { // The cached level ends here; the past-end signal serves drains. + // The frame stays claimed on every exit, valueless ones included. + claim_frame(content.layout()); return GPoll::Error(Box::new(core_types::gpoll::GraphError::past_end())); } if entry.generation == ctx.arena().generation() { // SAFETY: within the generation the materialized batch stays live, // immutable, and laid out at the recorded stride. - let rec = unsafe { core_types::record::Rec::new((entry.frames + lane * entry.stride) as *const u8) }; - return finalized(RecordValue::spilled(rec), &entry.finality); + let value = unsafe { serve_frame(content.layout(), (entry.frames + lane * entry.stride) as *const u8) }; + return finalized(value, &entry.finality); } match entry.lanes[lane].replay(content.layout(), ctx.arena()) { Some(value) => finalized(value, &entry.finality), @@ -97,8 +99,14 @@ fn memoize<'e>( *cache.lock().unwrap() = Some(entry); result } - LevelStatus::Pending => GPoll::Pending, - LevelStatus::Error(error) => GPoll::Error(Box::new(error)), + LevelStatus::Pending => { + claim_frame(content.layout()); + GPoll::Pending + } + LevelStatus::Error(error) => { + claim_frame(content.layout()); + GPoll::Error(Box::new(error)) + } }; } let result = content.eval(&ctx); @@ -143,20 +151,23 @@ fn frame_memo<'e>( }; // SAFETY: published bytes are same-frame copies of this edge's records, // so they carry the edge's layout with live parked references. - let revive = |bytes: &'e Box<[u8]>| unsafe { record_from_bytes(content.layout(), bytes) }; + let revive = |bytes: &'e Box<[u8]>| unsafe { serve_frame(content.layout(), bytes.as_ptr()) }; match table.lookup(cache_key(ctx)) { Lookup::Hit(Finality::AllFinal, bytes) => GPoll::Final(revive(bytes)), Lookup::Hit(Finality::Partial, bytes) => GPoll::Partial(revive(bytes)), Lookup::Vacant(slot) => match content.eval(&ctx) { GPoll::Final(value) => { // SAFETY: the value came from this edge, so it carries the edge's layout. + // The eval's own frame serves this pull; the publish feeds later ones. let bytes = unsafe { copy_record_bytes(content.layout(), content.layout().rec(&value)) }; - GPoll::Final(revive(slot.publish(bytes, Finality::AllFinal))) + slot.publish(bytes, Finality::AllFinal); + GPoll::Final(value) } GPoll::Partial(value) => { // SAFETY: as above. let bytes = unsafe { copy_record_bytes(content.layout(), content.layout().rec(&value)) }; - GPoll::Partial(revive(slot.publish(bytes, Finality::Partial))) + slot.publish(bytes, Finality::Partial); + GPoll::Partial(value) } unpublishable => { slot.release();