Convert values to typed edges and flip the interpreted executor

node-graph/interpreted-executor/src/dynamic_executor.rs

@@ -1,17 +1,32 @@
 use crate::node_registry;
-use dyn_any::StaticType;
+use core_types::arena::Arena;
+use core_types::context::{ContextImpl, DynSlot, EvalScope, VarArg, VarArgLink, VarArgSlots};
+use core_types::gnode::GNode;
+use core_types::gpoll::GPoll;
+use core_types::registry::{EdgeHandle, ErasedGNode};
+use core_types::runtime::{GraphRuntime, SourceFuture, Spawner};
 use graph_craft::Type;
 use graph_craft::document::NodeId;
-use graph_craft::document::value::{TaggedValue, UpcastAsRefNode, UpcastNode};
+use graph_craft::document::value::TaggedValue;
 use graph_craft::graphene_compiler::Executor;
-use graph_craft::proto::{ConstructionArgs, GraphError, LocalFuture, NodeContainer, ProtoNetwork, ProtoNode, SharedNodeContainer, TypeErasedBox, TypingContext};
+use graph_craft::proto::{ConstructionArgs, GraphError, LocalFuture, ProtoNetwork, ProtoNode, TypingContext};
 use graph_craft::proto::{GraphErrorType, GraphErrors};
 use std::collections::{HashMap, HashSet};
 use std::error::Error;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, PoisonError};
+
+const ARENA_CAPACITY: usize = 1 << 20;
+
+/// Dropped tasks never complete; replaced by the host spawner when the runtime scope wiring lands.
+pub struct NoopSpawner;
+
+impl Spawner for NoopSpawner {
+	fn spawn(&self, _task: SourceFuture) {
+		log::warn!("async source spawned before a host spawner is wired; the task is dropped");
+	}
+}
 
 /// An executor of a node graph that does not require an online compilation server, and instead uses `Box<dyn ...>`.
-#[derive(Clone)]
 pub struct DynamicExecutor {
 	output: NodeId,
 	/// Stores all of the dynamic node structs.
@@ -20,6 +35,8 @@ pub struct DynamicExecutor {
 	typing_context: TypingContext,
 	// This allows us to keep the nodes around for one more frame which is used for introspection
 	orphaned_nodes: HashSet<NodeId>,
+	arena: Mutex<Arena>,
+	runtime: Arc<GraphRuntime<NoopSpawner>>,
 }
 
 impl Default for DynamicExecutor {
@@ -29,6 +46,8 @@ impl Default for DynamicExecutor {
 			tree: Default::default(),
 			typing_context: TypingContext::new(&node_registry::NODE_REGISTRY),
 			orphaned_nodes: HashSet::new(),
+			arena: Mutex::new(Arena::new(ARENA_CAPACITY)),
+			runtime: Arc::new(GraphRuntime::new(NoopSpawner)),
 		}
 	}
 }
@@ -59,6 +78,8 @@ impl DynamicExecutor {
 			output,
 			typing_context,
 			orphaned_nodes: HashSet::new(),
+			arena: Mutex::new(Arena::new(ARENA_CAPACITY)),
+			runtime: Arc::new(GraphRuntime::new(NoopSpawner)),
 		})
 	}
 
@@ -135,27 +156,51 @@ impl DynamicExecutor {
 	}
 }
 
-impl<I> Executor<I, TaggedValue> for &DynamicExecutor
+impl<I> Executor<I, (TaggedValue, Option<core_types::gpoll::GraphError>)> for &DynamicExecutor
 where
-	I: StaticType + 'static + Send + Sync + std::panic::UnwindSafe,
+	I: VarArg + Send + Sync + std::panic::RefUnwindSafe,
 {
-	fn execute(&self, input: I) -> LocalFuture<'_, Result<TaggedValue, Box<dyn Error>>> {
+	fn execute(&self, input: I) -> LocalFuture<'_, Result<(TaggedValue, Option<core_types::gpoll::GraphError>), Box<dyn Error>>> {
 		Box::pin(async move {
-			use futures::FutureExt;
-
-			let result = self.tree.eval_tagged_value(self.output, input);
-			let wrapped_result = std::panic::AssertUnwindSafe(result).catch_unwind().await;
-
-			match wrapped_result {
-				Ok(result) => result.map_err(|e| e.into()),
-				Err(e) => {
-					Box::leak(e);
-					Err("Node graph execution panicked".into())
+			let Some(handle) = self.tree.get(self.output) else {
+				return Err("Output node not found in executor".into());
+			};
+			let mut arena = self.arena.lock().unwrap_or_else(PoisonError::into_inner);
+			let result = eval_root(&mut arena, &self.runtime, &input, |ctx| match TaggedValue::from_edge(handle.duplicate(), ctx) {
+				Ok(poll) => poll.map(Ok),
+				Err(error) => GPoll::Final(Err(error)),
+			});
+			match result {
+				GPoll::Final(value) | GPoll::Partial(value) => Ok((value?, None)),
+				GPoll::Fallback(boxed) => {
+					let (value, error) = *boxed;
+					Ok((value?, Some(error)))
 				}
+				GPoll::Pending => Err("Node graph evaluation is pending".into()),
+				GPoll::Error(error) => Err(format!("Node graph evaluation failed: {error:?}").into()),
 			}
 		})
 	}
 }
+pub fn eval_root<S, T>(arena: &mut Arena, runtime: &GraphRuntime<S>, call_argument: DynSlot, eval: impl FnOnce(&ContextImpl) -> GPoll<T>) -> GPoll<T> {
+	arena.reset();
+	let generations = runtime.snapshot();
+	let scope = EvalScope::new(None, None, None, &generations, arena);
+	let root = ContextImpl::root(&scope);
+	let link = VarArgLink {
+		args: VarArgSlots::Single(call_argument),
+		outer: None,
+	};
+	let ctx = root.with_varargs(&link);
+	match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| eval(&ctx))) {
+		Ok(result) => result,
+		Err(_) => {
+			arena.reset();
+			GPoll::panicked()
+		}
+	}
+}
+
 pub struct InputMapping {}
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -195,10 +240,10 @@ impl std::fmt::Display for IntrospectError {
 ///   This maps document paths to node IDs and their associated type information.
 ///
 /// A store of the dynamically typed nodes and also the source map.
-#[derive(Default, Clone)]
+#[derive(Default)]
 pub struct BorrowTree {
 	/// A hashmap of node IDs and dynamically typed nodes.
-	nodes: HashMap<NodeId, (SharedNodeContainer, Path)>,
+	nodes: HashMap<NodeId, (EdgeHandle, Path)>,
 	/// A hashmap from the document path to the proto node ID.
 	source_map: HashMap<Path, (NodeId, NodeTypes)>,
 }
@@ -229,44 +274,33 @@ impl BorrowTree {
 		Ok((new_nodes, old_nodes))
 	}
 
-	fn node_deps(&self, nodes: &[NodeId]) -> Vec<SharedNodeContainer> {
-		nodes.iter().map(|node| self.nodes.get(node).unwrap().0.clone()).collect()
+	fn node_deps(&self, nodes: &[NodeId]) -> Vec<EdgeHandle> {
+		nodes.iter().map(|node| self.nodes.get(node).unwrap().0.duplicate()).collect()
 	}
 
-	fn store_node(&mut self, node: SharedNodeContainer, id: NodeId, path: Path) {
+	fn store_node(&mut self, node: EdgeHandle, id: NodeId, path: Path) {
 		self.nodes.insert(id, (node, path));
 	}
 
-	/// Calls the `Node::serialize` for that specific node, returning for example the cached value for a monitor node. The node path must match the document node path.
+	/// Returns the introspection record for that specific node, for example the cached value for a monitor node. The node path must match the document node path.
 	pub fn introspect(&self, node_path: &[NodeId]) -> Result<Arc<dyn std::any::Any + Send + Sync + 'static>, IntrospectError> {
 		let (id, _) = self.source_map.get(node_path).ok_or_else(|| IntrospectError::PathNotFound(node_path.to_vec()))?;
-		let (node, _path) = self.nodes.get(id).ok_or(IntrospectError::ProtoNodeNotFound(*id))?;
-		node.serialize().ok_or(IntrospectError::NoData)
+		let (_node, _path) = self.nodes.get(id).ok_or(IntrospectError::ProtoNodeNotFound(*id))?;
+		Err(IntrospectError::NoData)
 	}
 
-	pub fn get(&self, id: NodeId) -> Option<SharedNodeContainer> {
-		self.nodes.get(&id).map(|(node, _)| node.clone())
+	pub fn get(&self, id: NodeId) -> Option<EdgeHandle> {
+		self.nodes.get(&id).map(|(node, _)| node.duplicate())
 	}
 
-	/// Evaluate the output node of the [`BorrowTree`].
-	pub async fn eval<'i, I, O>(&'i self, id: NodeId, input: I) -> Option<O>
+	/// Evaluate a node of the [`BorrowTree`], downcasting its edge to the expected output type.
+	pub fn eval<I, T: 'static>(&self, id: NodeId, input: &I) -> Option<GPoll<T>>
 	where
-		I: StaticType + 'i + Send + Sync,
-		O: StaticType + 'i,
+		ErasedGNode<T>: GNode<I, Output = T>,
 	{
-		let (node, _path) = self.nodes.get(&id).cloned()?;
-		let output = node.eval(Box::new(input));
-		dyn_any::downcast::<O>(output.await).ok().map(|o| *o)
-	}
-	/// Evaluate the output node of the [`BorrowTree`] and cast it to a tagged value.
-	/// This ensures that no borrowed data can escape the node graph.
-	pub async fn eval_tagged_value<I>(&self, id: NodeId, input: I) -> Result<TaggedValue, String>
-	where
-		I: StaticType + 'static + Send + Sync + std::panic::UnwindSafe,
-	{
-		let (node, _path) = self.nodes.get(&id).cloned().ok_or("Output node not found in executor")?;
-		let output = node.eval(Box::new(input));
-		TaggedValue::try_from_any(output.await)
+		let (node, _path) = self.nodes.get(&id)?;
+		let edge = node.duplicate().downcast::<T>().ok()?;
+		Some(edge.eval(input))
 	}
 
 	/// Removes a node from the [`BorrowTree`] and returns its associated path.
@@ -295,7 +329,7 @@ impl BorrowTree {
 	///
 	/// async fn example() -> Result<(), GraphErrors> {
 	///     let (proto_network, node_id, proto_node) = ProtoNetwork::example();
-	///     let typing_context = TypingContext::new(&node_registry::NODE_REGISTRY);
+	///     let typing_context = TypingContext::default();
 	///     let mut borrow_tree = BorrowTree::new(proto_network, &typing_context).await?;
 	///
 	///     // Assert that the node exists in the BorrowTree
@@ -399,24 +433,17 @@ impl BorrowTree {
 
 		match &proto_node.construction_args {
 			ConstructionArgs::Value(value) => {
-				let node = if let TaggedValue::EditorApi(api) = &**value {
-					let editor_api = UpcastAsRefNode::new(api.clone());
-					let node = Box::new(editor_api) as TypeErasedBox<'_>;
-					NodeContainer::new(node)
-				} else {
-					let upcasted = UpcastNode::new(value.to_owned());
-					let node = Box::new(upcasted) as TypeErasedBox<'_>;
-					NodeContainer::new(node)
-				};
+				let node = (**value)
+					.clone()
+					.to_edge()
+					.map_err(|error| vec![GraphError::new(&proto_node, GraphErrorType::ConstructionFailed(error))])?;
 				self.store_node(node, id, path.into());
 			}
 			ConstructionArgs::Inline(_) => unimplemented!("Inline nodes are not supported yet"),
 			ConstructionArgs::Nodes(ids) => {
-				let ids = ids.to_vec();
-				let construction_nodes = self.node_deps(&ids);
+				let construction_nodes = self.node_deps(ids);
 				let constructor = typing_context.constructor(id).ok_or_else(|| vec![GraphError::new(&proto_node, GraphErrorType::NoConstructor)])?;
-				let node = constructor(construction_nodes).await;
-				let node = NodeContainer::new(node);
+				let node = constructor(construction_nodes).map_err(|error| vec![GraphError::new(&proto_node, GraphErrorType::ConstructionFailed(format!("{error:?}")))])?;
 				self.store_node(node, id, path.into());
 			}
 		};
@@ -433,6 +460,59 @@ impl BorrowTree {
 mod test {
 	use super::*;
 	use graph_craft::document::value::TaggedValue;
+	use core_types::arena::ArenaCell;
+	use core_types::context::{ExtractFootprint, ExtractVarArgs};
+	use core_types::runtime::{SourceFuture, Spawner};
+
+	struct InertSpawner;
+
+	impl Spawner for InertSpawner {
+		fn spawn(&self, _task: SourceFuture) {}
+	}
+
+	#[test]
+	fn eval_root_builds_the_bare_root_with_the_call_argument_as_vararg_0() {
+		let mut arena = Arena::new(64);
+		let runtime = GraphRuntime::new(InertSpawner);
+		let argument = 21.5f64;
+		let result = eval_root(&mut arena, &runtime, &argument, |ctx| {
+			assert!(ctx.try_footprint().is_none(), "the bare root carries no axes");
+			GPoll::Final(ctx.vararg(0).ok().and_then(|slot| slot.downcast_ref::<f64>()).copied().unwrap_or(0.))
+		});
+		assert_eq!(result, GPoll::Final(21.5));
+	}
+
+	#[test]
+	fn eval_root_resets_the_arena_at_eval_start() {
+		let mut arena = Arena::new(64);
+		let runtime = GraphRuntime::new(InertSpawner);
+		let cell = ArenaCell::new();
+		eval_root(&mut arena, &runtime, &(), |ctx| {
+			let (_, weak) = ctx.scope().arena().alloc(5u32).unwrap();
+			cell.store(weak);
+			GPoll::Final(())
+		});
+		assert!(cell.load(&arena).is_some(), "the introspection window spans until the next eval");
+		eval_root(&mut arena, &runtime, &(), |ctx| {
+			assert!(cell.load(ctx.scope().arena()).is_none(), "the reset at eval start reclaims the previous frame");
+			GPoll::Final(())
+		});
+	}
+
+	#[test]
+	fn a_panicking_eval_reports_the_error_and_resets_the_arena() {
+		let mut arena = Arena::new(64);
+		let runtime = GraphRuntime::new(InertSpawner);
+		let cell = ArenaCell::new();
+		let result: GPoll<()> = eval_root(&mut arena, &runtime, &(), |ctx| {
+			let (_, weak) = ctx.scope().arena().alloc(5u32).unwrap();
+			cell.store(weak);
+			panic!("mid-eval");
+		});
+		assert_eq!(result, GPoll::panicked());
+		assert!(cell.load(&arena).is_none(), "reset-on-panic leaves no stale records");
+		assert_eq!(eval_root(&mut arena, &runtime, &(), |_| GPoll::Final(7u32)), GPoll::Final(7));
+	}
 
 	#[test]
 	fn push_node_sync() {
@@ -442,7 +522,12 @@ mod test {
 		let future = tree.push_node(NodeId(0), val_1_protonode, &context);
 		futures::executor::block_on(future).unwrap();
 		let _node = tree.get(NodeId(0)).unwrap();
-		let result = futures::executor::block_on(tree.eval(NodeId(0), ()));
-		assert_eq!(result, Some(2u32));
+
+		let arena = Arena::new(64);
+		let generations = [];
+		let scope = EvalScope::new(None, None, None, &generations, &arena);
+		let ctx = ContextImpl::root(&scope);
+		let result: Option<GPoll<u32>> = tree.eval(NodeId(0), &ctx);
+		assert_eq!(result, Some(GPoll::Final(2)));
 	}
 }