diff --git a/node-graph/libraries/core-types/src/context.rs b/node-graph/libraries/core-types/src/context.rs index 01a6574723..d0ba3778ca 100644 --- a/node-graph/libraries/core-types/src/context.rs +++ b/node-graph/libraries/core-types/src/context.rs @@ -849,6 +849,78 @@ impl PositionScope<'_, C> { } } +#[derive(Clone, Debug, Default)] +pub struct CtxSnapshot { + footprint: Option, + real_time: Option, + animation_time: Option, + pointer_position: Option, + index: Vec, + positions: Vec, + generations: Vec<(SourceId, u64)>, +} + +impl CtxSnapshot { + pub fn capture(ctx: &C) -> Self + where + C: DeriveCtx + ExtractFootprint + ExtractRealTime + ExtractAnimationTime + ExtractPointerPosition + ExtractIndex + ExtractPosition, + { + Self { + footprint: ctx.try_footprint().copied(), + real_time: ctx.try_real_time(), + animation_time: ctx.try_animation_time(), + pointer_position: ctx.try_pointer_position(), + index: ctx.try_index().map(|levels| levels.collect()).unwrap_or_default(), + positions: ctx.try_position().map(|positions| positions.collect()).unwrap_or_default(), + generations: ctx.scope().generations().to_vec(), + } + } + + pub fn generations(&self) -> &[(SourceId, u64)] { + &self.generations + } + + pub fn scope<'s>(&'s self, arena: &'s Arena) -> EvalScope<'s> { + EvalScope::new(self.real_time, self.animation_time, self.pointer_position, &self.generations, arena) + } +} + +impl ExtractFootprint for CtxSnapshot { + fn try_footprint(&self) -> Option<&Footprint> { + self.footprint.as_ref() + } +} + +impl ExtractRealTime for CtxSnapshot { + fn try_real_time(&self) -> Option { + self.real_time + } +} + +impl ExtractAnimationTime for CtxSnapshot { + fn try_animation_time(&self) -> Option { + self.animation_time + } +} + +impl ExtractPointerPosition for CtxSnapshot { + fn try_pointer_position(&self) -> Option { + self.pointer_position + } +} + +impl ExtractIndex for CtxSnapshot { + fn try_index(&self) -> Option> { + Some(self.index.iter().copied()) + } +} + +impl ExtractPosition for CtxSnapshot { + fn try_position(&self) -> Option> { + Some(self.positions.iter().copied()) + } +} + pub struct VarArgScope<'c, C> { ctx: &'c C, link: VarArgLink<'c>, diff --git a/node-graph/libraries/core-types/src/gnode.rs b/node-graph/libraries/core-types/src/gnode.rs index 7bf335b02a..4a65b36554 100644 --- a/node-graph/libraries/core-types/src/gnode.rs +++ b/node-graph/libraries/core-types/src/gnode.rs @@ -180,6 +180,7 @@ impl StatusCell { pub fn merge(self, poll: GPoll) -> GPoll { match poll { GPoll::Final(value) => self.finish(value), + GPoll::Partial(_) if self.no_partial => GPoll::Pending, GPoll::Partial(value) => match self.finish(value) { GPoll::Final(value) => GPoll::Partial(value), other => other, diff --git a/node-graph/libraries/core-types/src/lib.rs b/node-graph/libraries/core-types/src/lib.rs index 3036ddcfb8..610d4cff04 100644 --- a/node-graph/libraries/core-types/src/lib.rs +++ b/node-graph/libraries/core-types/src/lib.rs @@ -15,6 +15,7 @@ pub mod misc; pub mod ops; pub mod registry; pub mod render_complexity; +pub mod runtime; pub mod transform; pub mod uuid; pub mod value; diff --git a/node-graph/libraries/core-types/src/runtime.rs b/node-graph/libraries/core-types/src/runtime.rs new file mode 100644 index 0000000000..e12cb77a90 --- /dev/null +++ b/node-graph/libraries/core-types/src/runtime.rs @@ -0,0 +1,176 @@ +use crate::SourceId; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; + +#[cfg(not(target_family = "wasm"))] +pub type SourceFuture = Pin + Send + 'static>>; +#[cfg(target_family = "wasm")] +pub type SourceFuture = Pin + 'static>>; + +#[cfg(not(target_family = "wasm"))] +pub type DynRuntime = dyn Runtime + Send + Sync; +#[cfg(target_family = "wasm")] +pub type DynRuntime = dyn Runtime; + +pub trait Runtime { + fn spawn(&self, source: SourceId, future: SourceFuture); +} + +#[derive(Clone)] +pub struct RuntimeHandle(pub Arc); + +impl std::fmt::Debug for RuntimeHandle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RuntimeHandle").finish_non_exhaustive() + } +} + +impl graphene_hash::CacheHash for RuntimeHandle { + fn cache_hash(&self, _state: &mut H) {} +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::arena::Arena; + use crate::context::{ContextImpl, Ctx, CtxSnapshot, EvalScope, ExtractFootprint}; + use crate::gnode::GNode; + use crate::gpoll::GPoll; + use crate::transform::Footprint; + use std::sync::Mutex; + use std::sync::atomic::{AtomicU32, Ordering}; + + #[derive(Default)] + struct MockRuntime { + futures: Mutex>, + } + + impl Runtime for MockRuntime { + fn spawn(&self, source: SourceId, future: SourceFuture) { + self.futures.lock().unwrap().push((source, future)); + } + } + + impl MockRuntime { + fn drain(&self) -> Vec { + let futures = std::mem::take(&mut *self.futures.lock().unwrap()); + let mut task_ctx = std::task::Context::from_waker(std::task::Waker::noop()); + futures + .into_iter() + .map(|(source, mut future)| { + assert!(future.as_mut().poll(&mut task_ctx).is_ready()); + source + }) + .collect() + } + } + + struct SourceNode(T); + + impl GNode for SourceNode { + type Output = T; + + fn eval(&self, _input: &Input) -> GPoll { + GPoll::Final(self.0.clone()) + } + } + + static SLOW_DOUBLE_RUNS: AtomicU32 = AtomicU32::new(0); + + #[node_macro::node(category(""))] + async fn slow_double(_: impl Ctx, value: f64) -> f64 { + SLOW_DOUBLE_RUNS.fetch_add(1, Ordering::Relaxed); + value * 2. + } + + fn stand_in(_value: &f64) -> f64 { + -1. + } + + #[node_macro::node(category(""), placeholder(stand_in))] + async fn preview_double(_: impl Ctx, value: f64) -> f64 { + value * 2. + } + + #[node_macro::node(category(""), placeholder(stand_in), no_partial)] + async fn strict_double(_: impl Ctx, value: f64) -> f64 { + value * 2. + } + + #[node_macro::node(category(""))] + async fn snapshot_resolution(ctx: CtxSnapshot, _primary: ()) -> u32 { + ctx.try_footprint().map(|footprint| footprint.resolution.x).unwrap_or(0) + } + + fn scope_fixture<'a>(generations: &'a [(SourceId, u64)], arena: &'a Arena) -> EvalScope<'a> { + EvalScope::new(None, None, None, generations, arena) + } + + #[test] + fn async_source_spawns_once_and_lands_via_the_slot() { + let arena = Arena::new(64); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let ctx = ContextImpl::root(&scope); + + let runtime = Arc::new(MockRuntime::default()); + let graph = SlowDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(7u64)); + + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Pending); + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Pending); + assert_eq!(SLOW_DOUBLE_RUNS.load(Ordering::Relaxed), 0); + assert_eq!(runtime.drain(), vec![7]); + assert_eq!(SLOW_DOUBLE_RUNS.load(Ordering::Relaxed), 1); + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Final(42.0)); + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Final(42.0)); + assert_eq!(SLOW_DOUBLE_RUNS.load(Ordering::Relaxed), 1); + } + + #[test] + fn async_source_reports_the_placeholder_while_in_flight() { + let arena = Arena::new(64); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let ctx = ContextImpl::root(&scope); + + let runtime = Arc::new(MockRuntime::default()); + let graph = PreviewDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(1u64)); + + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Partial(-1.0)); + runtime.drain(); + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Final(42.0)); + } + + #[test] + fn no_partial_maps_the_placeholder_frame_to_pending() { + let arena = Arena::new(64); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let ctx = ContextImpl::root(&scope); + + let runtime = Arc::new(MockRuntime::default()); + let graph = StrictDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(2u64)); + + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Pending); + runtime.drain(); + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Final(42.0)); + } + + #[test] + fn async_kernels_read_the_captured_context_snapshot() { + let arena = Arena::new(64); + let generations = []; + let scope = scope_fixture(&generations, &arena); + let root = ContextImpl::root(&scope); + let footprint = Footprint::DEFAULT; + let ctx = root.with_footprint(&footprint); + + let runtime = Arc::new(MockRuntime::default()); + let graph = SnapshotResolutionNode::new(SourceNode(()), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(3u64)); + + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Pending); + runtime.drain(); + assert_eq!(GNode::eval(&graph, &ctx), GPoll::Final(Footprint::DEFAULT.resolution.x)); + } +}