Add the async source runtime vocabulary and context snapshot

node-graph/libraries/core-types/src/context.rs

@@ -849,6 +849,78 @@ impl<C: DeriveCtx> PositionScope<'_, C> {
 	}
 }
 
+#[derive(Clone, Debug, Default)]
+pub struct CtxSnapshot {
+	footprint: Option<Footprint>,
+	real_time: Option<f64>,
+	animation_time: Option<f64>,
+	pointer_position: Option<DVec2>,
+	index: Vec<usize>,
+	positions: Vec<DVec2>,
+	generations: Vec<(SourceId, u64)>,
+}
+
+impl CtxSnapshot {
+	pub fn capture<C>(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<f64> {
+		self.real_time
+	}
+}
+
+impl ExtractAnimationTime for CtxSnapshot {
+	fn try_animation_time(&self) -> Option<f64> {
+		self.animation_time
+	}
+}
+
+impl ExtractPointerPosition for CtxSnapshot {
+	fn try_pointer_position(&self) -> Option<DVec2> {
+		self.pointer_position
+	}
+}
+
+impl ExtractIndex for CtxSnapshot {
+	fn try_index(&self) -> Option<impl Iterator<Item = usize>> {
+		Some(self.index.iter().copied())
+	}
+}
+
+impl ExtractPosition for CtxSnapshot {
+	fn try_position(&self) -> Option<impl Iterator<Item = DVec2>> {
+		Some(self.positions.iter().copied())
+	}
+}
+
 pub struct VarArgScope<'c, C> {
 	ctx: &'c C,
 	link: VarArgLink<'c>,

node-graph/libraries/core-types/src/gnode.rs

@@ -180,6 +180,7 @@ impl StatusCell {
 	pub fn merge<T>(self, poll: GPoll<T>) -> GPoll<T> {
 		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,

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;

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<Box<dyn Future<Output = ()> + Send + 'static>>;
+#[cfg(target_family = "wasm")]
+pub type SourceFuture = Pin<Box<dyn Future<Output = ()> + '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<DynRuntime>);
+
+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<H: std::hash::Hasher>(&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<Vec<(SourceId, SourceFuture)>>,
+	}
+
+	impl Runtime for MockRuntime {
+		fn spawn(&self, source: SourceId, future: SourceFuture) {
+			self.futures.lock().unwrap().push((source, future));
+		}
+	}
+
+	impl MockRuntime {
+		fn drain(&self) -> Vec<SourceId> {
+			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>(T);
+
+	impl<T: Clone, Input> GNode<Input> for SourceNode<T> {
+		type Output = T;
+
+		fn eval(&self, _input: &Input) -> GPoll<T> {
+			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));
+	}
+}