Claim the node frame on every cache serve and reclaim extracted edges

This commit is contained in:
Dennis Kobert
2026-08-23 08:31:39 +00:00
parent 58182bc2d3
commit e40ee486a9
2 changed files with 74 additions and 23 deletions
+55 -15
View File
@@ -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() unsafe { std::slice::from_raw_parts(rec.ptr(), layout.size) }.into()
} }
/// Builds a record value over `bytes`, a record of `layout` copied out /// Serves a record whose frame lives in storage the current evaluation does
/// earlier: inline layouts copy into the value, spilled ones alias the /// not own (a cached batch, published bytes): claims the node's frame over a
/// bytes. /// 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 /// # Safety
/// `bytes` must hold a record of `layout` whose parked references are still /// `src` must point at a live record of `layout` whose parked references
/// live; both hold for a copy taken in the same evaluation frame. /// outlive the serving evaluation.
pub unsafe fn record_from_bytes<'e>(layout: &Layout, bytes: &'e [u8]) -> RecordValue<'e> { pub unsafe fn serve_frame<'e>(layout: &Layout, src: *const u8) -> RecordValue<'e> {
if layout.frame_bytes() == 0 { if layout.frame_bytes() == 0 {
let mut value = RecordValue::zeroed(); 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 value
} else { } 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 /// A captured record: the layout plus a generation-checked handle to the
/// arena copy, materialized by the introspection holder, which owns the /// arena copy, materialized by the introspection holder, which owns the
/// arena. A dead generation materializes to `None`, never to a stale read. /// arena. A dead generation materializes to `None`, never to a stale read.
@@ -1700,7 +1724,12 @@ where
type Output = El; type Output = El;
fn eval(&self, input: &C) -> GPoll<El> { fn eval(&self, input: &C) -> GPoll<El> {
self.edge.eval(input).map(|value| unsafe { read_element::<El>(self.layout.rec(&value)) }) let mark = stack::sp();
let result = self.edge.eval(input).map(|value| unsafe { read_element::<El>(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<RecordValue<'e>> { fn eval(&self, input: &C) -> GPoll<RecordValue<'e>> {
match &self.plan { match &self.plan {
None => self.edge.eval(input), None => self.edge.eval(input),
Some(plan) if plan.union.frame_bytes() == 0 => self.edge.eval(input).map(|value| { Some(plan) if plan.union.frame_bytes() == 0 => {
let mut out = RecordValue::zeroed(); let mark = stack::sp();
unsafe { plan.translate(plan.source.rec(&value), out.as_mut_ptr()) }; let result = self.edge.eval(input).map(|value| {
out 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) => { Some(plan) => {
let dst = stack::push(plan.union.frame_bytes()); let dst = stack::push(plan.union.frame_bytes());
let value = self.edge.eval(input); 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
} }
} }
} }
+19 -8
View File
@@ -4,7 +4,7 @@ use core_types::frame_table::{FrameTable, Lookup};
use core_types::gpoll::{Finality, GPoll}; use core_types::gpoll::{Finality, GPoll};
use core_types::graphene_hash::CacheHash; use core_types::graphene_hash::CacheHash;
use core_types::memo::IORecord; 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 core_types::registry::cache_key;
use std::sync::Arc; use std::sync::Arc;
use std::sync::Mutex; use std::sync::Mutex;
@@ -58,13 +58,15 @@ fn memoize<'e>(
let serve = |entry: &MemoLevel| { let serve = |entry: &MemoLevel| {
if lane >= entry.lanes.len() { if lane >= entry.lanes.len() {
// The cached level ends here; the past-end signal serves drains. // 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())); return GPoll::Error(Box::new(core_types::gpoll::GraphError::past_end()));
} }
if entry.generation == ctx.arena().generation() { if entry.generation == ctx.arena().generation() {
// SAFETY: within the generation the materialized batch stays live, // SAFETY: within the generation the materialized batch stays live,
// immutable, and laid out at the recorded stride. // immutable, and laid out at the recorded stride.
let rec = unsafe { core_types::record::Rec::new((entry.frames + lane * entry.stride) as *const u8) }; let value = unsafe { serve_frame(content.layout(), (entry.frames + lane * entry.stride) as *const u8) };
return finalized(RecordValue::spilled(rec), &entry.finality); return finalized(value, &entry.finality);
} }
match entry.lanes[lane].replay(content.layout(), ctx.arena()) { match entry.lanes[lane].replay(content.layout(), ctx.arena()) {
Some(value) => finalized(value, &entry.finality), Some(value) => finalized(value, &entry.finality),
@@ -97,8 +99,14 @@ fn memoize<'e>(
*cache.lock().unwrap() = Some(entry); *cache.lock().unwrap() = Some(entry);
result result
} }
LevelStatus::Pending => GPoll::Pending, LevelStatus::Pending => {
LevelStatus::Error(error) => GPoll::Error(Box::new(error)), claim_frame(content.layout());
GPoll::Pending
}
LevelStatus::Error(error) => {
claim_frame(content.layout());
GPoll::Error(Box::new(error))
}
}; };
} }
let result = content.eval(&ctx); 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, // SAFETY: published bytes are same-frame copies of this edge's records,
// so they carry the edge's layout with live parked references. // 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)) { match table.lookup(cache_key(ctx)) {
Lookup::Hit(Finality::AllFinal, bytes) => GPoll::Final(revive(bytes)), Lookup::Hit(Finality::AllFinal, bytes) => GPoll::Final(revive(bytes)),
Lookup::Hit(Finality::Partial, bytes) => GPoll::Partial(revive(bytes)), Lookup::Hit(Finality::Partial, bytes) => GPoll::Partial(revive(bytes)),
Lookup::Vacant(slot) => match content.eval(&ctx) { Lookup::Vacant(slot) => match content.eval(&ctx) {
GPoll::Final(value) => { GPoll::Final(value) => {
// SAFETY: the value came from this edge, so it carries the edge's layout. // 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)) }; 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) => { GPoll::Partial(value) => {
// SAFETY: as above. // SAFETY: as above.
let bytes = unsafe { copy_record_bytes(content.layout(), content.layout().rec(&value)) }; 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 => { unpublishable => {
slot.release(); slot.release();