From 167d733f0259a447bf5a6d1ffcc785e590ba59d4 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Fri, 31 Jul 2026 13:30:42 +0000 Subject: [PATCH 1/4] Wire the async source runtimes into the hosts --- desktop/src/app.rs | 5 ++ desktop/wrapper/src/lib.rs | 4 ++ editor/src/node_graph_executor.rs | 16 +++++- editor/src/node_graph_executor/runtime.rs | 70 ++++++++++++++++++++++- node-graph/graphene-cli/src/export.rs | 34 +++++++---- node-graph/graphene-cli/src/main.rs | 35 ++++++++++-- 6 files changed, 146 insertions(+), 18 deletions(-) diff --git a/desktop/src/app.rs b/desktop/src/app.rs index cc20d3abcd..887e19fb88 100644 --- a/desktop/src/app.rs +++ b/desktop/src/app.rs @@ -97,6 +97,11 @@ impl App { }); let desktop_wrapper = DesktopWrapper::new(rand::rng().random(), Arc::new(resource_storage), dirs::app_autosave_documents_dir(), wgpu_context.clone(), wake); + let completion_render_sender = start_render_sender.clone(); + DesktopWrapper::set_completion_notifier(move || { + let _ = completion_render_sender.try_send(()); + }); + Self { render_state: None, wgpu_context, diff --git a/desktop/wrapper/src/lib.rs b/desktop/wrapper/src/lib.rs index 5f2adff705..ff89db7636 100644 --- a/desktop/wrapper/src/lib.rs +++ b/desktop/wrapper/src/lib.rs @@ -52,6 +52,10 @@ impl DesktopWrapper { executor.execute() } + pub fn set_completion_notifier(notifier: impl Fn() + Send + Sync + 'static) { + graphite_editor::node_graph_executor::set_completion_notifier(Arc::new(notifier)); + } + pub async fn execute_node_graph() -> NodeGraphExecutionResult { let result = graphite_editor::node_graph_executor::run_node_graph().await; match result { diff --git a/editor/src/node_graph_executor.rs b/editor/src/node_graph_executor.rs index 348feb9c05..2cbc23297a 100644 --- a/editor/src/node_graph_executor.rs +++ b/editor/src/node_graph_executor.rs @@ -59,6 +59,9 @@ pub struct NodeGraphExecutor { runtime_io: NodeRuntimeIO, current_execution_id: u64, futures: VecDeque<(u64, ExecutionContext)>, + /// The most recently consumed plain render execution, kept so a runtime-replayed response with the same id + /// (sent after an async source completion) finds its context again. + last_execution_context: Option<(u64, ExecutionContext)>, node_graph_hash: u64, /// Full path from the root document network to the node currently being inspected by the Data panel, or empty if nothing is selected. /// The last element is the inspect target itself; preceding elements identify the nested subnetwork the node lives in, @@ -108,6 +111,7 @@ impl NodeGraphExecutor { let node_executor = Self { futures: Default::default(), runtime_io: NodeRuntimeIO::with_channels(request_sender, response_receiver), + last_execution_context: None, node_graph_hash: 0, current_execution_id: 0, previous_node_to_inspect: Vec::new(), @@ -377,9 +381,19 @@ impl NodeGraphExecutor { let execution_context = if self.futures.front().is_some_and(|&(queued_execution_id, _)| queued_execution_id == execution_id) { let (_, execution_context) = self.futures.pop_front().expect("front was just matched"); + self.last_execution_context = Some((execution_id, execution_context.clone())); execution_context } else { - panic!("InvalidGenerationId") + // A runtime-replayed response re-uses an already consumed id; only plain renders may re-apply. + match &self.last_execution_context { + Some((last_execution_id, execution_context)) if *last_execution_id == execution_id => { + if execution_context.export_config.is_some() || execution_context.measure_fill.is_some() { + continue; + } + execution_context.clone() + } + _ => panic!("InvalidGenerationId"), + } }; // TODO: Eventually remove this document upgrade code diff --git a/editor/src/node_graph_executor/runtime.rs b/editor/src/node_graph_executor/runtime.rs index eb2e6fade7..243bdb9e88 100644 --- a/editor/src/node_graph_executor/runtime.rs +++ b/editor/src/node_graph_executor/runtime.rs @@ -17,7 +17,7 @@ use graphene_std::ops::ConvertAsync; use graphene_std::platform_application_io::canvas_utils::{Canvas, CanvasSurface, CanvasSurfaceHandle}; use graphene_std::raster_types::Raster; use graphene_std::renderer::{Render, RenderParams, RenderSvgSegmentList, SvgRender, SvgSegment}; -use graphene_std::runtime::{DynGraphRuntime, DynSpawner, GraphRuntime, NoopSpawner, RuntimeHandle}; +use graphene_std::runtime::{DynGraphRuntime, DynNotifier, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner}; use graphene_std::transform::RenderQuality; use graphene_std::vector::Vector; use graphene_std::vector::style::RenderMode; @@ -43,6 +43,8 @@ pub struct NodeRuntime { update_thumbnails: bool, #[expect(dead_code, reason = "read once the host wires a notifier onto the runtime")] graph_runtime: Arc, + /// The last plain render request, replayed when an async source completion marks the graph dirty. + last_render: Option, editor_api: Arc, resources: ResourceRegistry, @@ -122,9 +124,57 @@ impl NodeGraphUpdateSender for InternalNodeGraphUpdateSender { // TODO: Replace with `core::cell::LazyCell` () or similar pub static NODE_RUNTIME: once_cell::sync::Lazy>> = once_cell::sync::Lazy::new(|| Mutex::new(None)); +#[cfg(not(target_family = "wasm"))] +pub struct TokioSpawner(Option); + +#[cfg(not(target_family = "wasm"))] +impl TokioSpawner { + pub fn new() -> Self { + Self(Some(tokio::runtime::Runtime::new().expect("Failed to start the async source runtime"))) + } +} + +#[cfg(not(target_family = "wasm"))] +impl Default for TokioSpawner { + fn default() -> Self { + Self::new() + } +} + +#[cfg(not(target_family = "wasm"))] +impl Spawner for TokioSpawner { + fn spawn(&self, task: SourceFuture) { + self.0.as_ref().expect("runtime lives until drop").spawn(task); + } +} + +/// Dropping a tokio runtime blocks on its tasks, which panics inside an async context; the tests drop +/// [`NodeRuntime`] from one, so shut down in the background instead. +#[cfg(not(target_family = "wasm"))] +impl Drop for TokioSpawner { + fn drop(&mut self) { + if let Some(runtime) = self.0.take() { + runtime.shutdown_background(); + } + } +} + +#[cfg(target_family = "wasm")] +pub struct WasmSpawner; + +#[cfg(target_family = "wasm")] +impl Spawner for WasmSpawner { + fn spawn(&self, task: SourceFuture) { + wasm_bindgen_futures::spawn_local(task); + } +} + impl NodeRuntime { pub fn new(receiver: Receiver, sender: Sender) -> Self { - let spawner: Box = Box::new(NoopSpawner); + #[cfg(not(target_family = "wasm"))] + let spawner: Box = Box::new(TokioSpawner::new()); + #[cfg(target_family = "wasm")] + let spawner: Box = Box::new(WasmSpawner); let graph_runtime: Arc = Arc::new(GraphRuntime::new(spawner)); let mut executor = DynamicExecutor::default(); executor.set_runtime(Arc::clone(&graph_runtime)); @@ -138,6 +188,7 @@ impl NodeRuntime { resources: ResourceRegistry::default(), update_thumbnails: true, graph_runtime: Arc::clone(&graph_runtime), + last_render: None, editor_api: PlatformEditorApi { editor_preferences: Box::new(EditorPreferences::default()), @@ -188,6 +239,10 @@ impl NodeRuntime { } let for_export = execution_request.render_config.for_export; + if !for_export { + self.last_render = Some(execution_request.clone()); + } + execution = Some(request); // If we get an export request we always execute it immedeatly otherwise it could get deduplicated @@ -207,6 +262,10 @@ impl NodeRuntime { eyedropper.render_config.pointer = execution.render_config.pointer; } + if self.executor.take_dirty() && execution.is_none() { + execution = self.last_render.clone().map(GraphRuntimeRequest::ExecutionRequest); + } + let requests = [preferences, graph, eyedropper, execution].into_iter().flatten(); for request in requests { @@ -582,6 +641,13 @@ pub(crate) fn replace_application_io(application_io: PlatformApplicationIo) { } } +pub fn set_completion_notifier(notifier: Arc) { + let node_runtime = NODE_RUNTIME.lock(); + if let Some(node_runtime) = &*node_runtime { + node_runtime.graph_runtime.set_notifier(notifier); + } +} + impl NodeRuntime { pub(crate) fn replace_application_io(&mut self, application_io: PlatformApplicationIo) { self.editor_api = PlatformEditorApi { diff --git a/node-graph/graphene-cli/src/export.rs b/node-graph/graphene-cli/src/export.rs index ba414fa54c..68a1120a52 100644 --- a/node-graph/graphene-cli/src/export.rs +++ b/node-graph/graphene-cli/src/export.rs @@ -10,18 +10,28 @@ use interpreted_executor::dynamic_executor::DynamicExecutor; use std::error::Error; use std::io::Cursor; use std::path::{Path, PathBuf}; +use std::sync::mpsc::Receiver; use std::time::Duration; -fn execute_to_final(executor: &DynamicExecutor, render_config: RenderConfig) -> Result> { - match executor.execute(render_config)? { - GPoll::Final(value) => Ok(value), - GPoll::Fallback(boxed) => { - let (value, error) = *boxed; - log::warn!("Node graph evaluation reported an error alongside its fallback output: {error:?}"); - Ok(value) +const SOURCE_COMPLETION_TIMEOUT: Duration = Duration::from_secs(30); + +fn execute_until_final(executor: &DynamicExecutor, render_config: RenderConfig, completion: &Receiver<()>) -> Result> { + loop { + while completion.try_recv().is_ok() {} + match executor.execute(render_config)? { + GPoll::Final(value) => return Ok(value), + GPoll::Fallback(boxed) => { + let (value, error) = *boxed; + log::warn!("Node graph evaluation reported an error alongside its fallback output: {error:?}"); + return Ok(value); + } + GPoll::Partial(_) | GPoll::Pending => { + completion + .recv_timeout(SOURCE_COMPLETION_TIMEOUT) + .map_err(|_| format!("Timed out after {}s waiting for async sources to complete", SOURCE_COMPLETION_TIMEOUT.as_secs()))?; + } + GPoll::Error(error) => return Err(format!("Node graph evaluation failed: {error:?}").into()), } - GPoll::Partial(_) | GPoll::Pending => Err("Node graph evaluation did not complete".into()), - GPoll::Error(error) => Err(format!("Node graph evaluation failed: {error:?}").into()), } } @@ -52,6 +62,7 @@ pub fn export_document( scale: f64, (width, height): (Option, Option), transparent: bool, + completion: &Receiver<()>, ) -> Result<(), Box> { // Determine export format based on file type let export_format = match file_type { @@ -73,7 +84,7 @@ pub fn export_document( } // Execute the graph - let result = execute_to_final(executor, render_config)?; + let result = execute_until_final(executor, render_config, completion)?; // Handle the result based on output type match result { @@ -172,6 +183,7 @@ pub fn export_gif( scale: f64, (width, height): (Option, Option), animation: AnimationParams, + completion: &Receiver<()>, ) -> Result<(), Box> { use image::codecs::gif::{GifEncoder, Repeat}; use image::{Frame, RgbaImage}; @@ -211,7 +223,7 @@ pub fn export_gif( } // Execute the graph for this frame - let result = execute_to_final(executor, render_config)?; + let result = execute_until_final(executor, render_config, completion)?; // Extract RGBA data from result let (data, img_width, img_height) = match result { diff --git a/node-graph/graphene-cli/src/main.rs b/node-graph/graphene-cli/src/main.rs index 746fb6ab4d..77203880f7 100644 --- a/node-graph/graphene-cli/src/main.rs +++ b/node-graph/graphene-cli/src/main.rs @@ -13,7 +13,7 @@ use graph_craft::graphene_compiler::Compiler; use graph_craft::proto::ProtoNetwork; use graph_craft::util::load_network; use graphene_std::application_io::{ApplicationIo, NodeGraphUpdateMessage, NodeGraphUpdateSender}; -use graphene_std::runtime::{DynGraphRuntime, DynSpawner, GraphRuntime, NoopSpawner, RuntimeHandle}; +use graphene_std::runtime::{DynGraphRuntime, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner}; use interpreted_executor::dynamic_executor::DynamicExecutor; use interpreted_executor::util::wrap_network_in_scope; use std::error::Error; @@ -28,6 +28,29 @@ impl NodeGraphUpdateSender for UpdateLogger { } } +struct TokioSpawner(Option); + +impl TokioSpawner { + fn new() -> Self { + Self(Some(tokio::runtime::Runtime::new().expect("Failed to start the async source runtime"))) + } +} + +impl Spawner for TokioSpawner { + fn spawn(&self, task: SourceFuture) { + self.0.as_ref().expect("runtime lives until drop").spawn(task); + } +} + +/// Dropping a tokio runtime blocks on its tasks, which panics inside the async main; shut down in the background instead. +impl Drop for TokioSpawner { + fn drop(&mut self) { + if let Some(runtime) = self.0.take() { + runtime.shutdown_background(); + } + } +} + #[derive(Debug, Parser)] #[clap(name = "graphene-cli", version)] pub struct App { @@ -179,7 +202,11 @@ fn main() -> Result<(), Box> { let preferences = EditorPreferences { max_render_region_size: EditorPreferences::default().max_render_region_size, }; - let graph_runtime: Arc = Arc::new(GraphRuntime::new(Box::new(NoopSpawner) as Box)); + let graph_runtime: Arc = Arc::new(GraphRuntime::new(Box::new(TokioSpawner::new()) as Box)); + let (completion_sender, completion_receiver) = std::sync::mpsc::channel(); + graph_runtime.set_notifier(Arc::new(move || { + let _ = completion_sender.send(()); + })); let editor_api = Arc::new(PlatformEditorApi { application_io: Some(application_io_for_api), node_graph_message_sender: Box::new(UpdateLogger {}), @@ -226,9 +253,9 @@ fn main() -> Result<(), Box> { // Perform export based on file type if file_type == export::FileType::Gif { let animation = export::AnimationParams::new(fps, frames, duration); - export::export_gif(&executor, wgpu_executor_ref.clone(), output, scale, (width, height), animation)?; + export::export_gif(&executor, wgpu_executor_ref.clone(), output, scale, (width, height), animation, &completion_receiver)?; } else { - export::export_document(&executor, wgpu_executor_ref.clone(), output, file_type, scale, (width, height), transparent)?; + export::export_document(&executor, wgpu_executor_ref.clone(), output, file_type, scale, (width, height), transparent, &completion_receiver)?; } } _ => unreachable!("All other commands should be handled before this match statement is run"), From 2b3161773fd1e757e08102b84f72e56535e86af9 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Sun, 2 Aug 2026 19:29:33 +0200 Subject: [PATCH 2/4] Propagate tokio runtime creation failure in graphene-cli --- node-graph/graphene-cli/src/main.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/node-graph/graphene-cli/src/main.rs b/node-graph/graphene-cli/src/main.rs index 77203880f7..81311bdda6 100644 --- a/node-graph/graphene-cli/src/main.rs +++ b/node-graph/graphene-cli/src/main.rs @@ -31,8 +31,8 @@ impl NodeGraphUpdateSender for UpdateLogger { struct TokioSpawner(Option); impl TokioSpawner { - fn new() -> Self { - Self(Some(tokio::runtime::Runtime::new().expect("Failed to start the async source runtime"))) + fn new() -> Result { + Ok(Self(Some(tokio::runtime::Runtime::new()?))) } } @@ -202,7 +202,7 @@ fn main() -> Result<(), Box> { let preferences = EditorPreferences { max_render_region_size: EditorPreferences::default().max_render_region_size, }; - let graph_runtime: Arc = Arc::new(GraphRuntime::new(Box::new(TokioSpawner::new()) as Box)); + let graph_runtime: Arc = Arc::new(GraphRuntime::new(Box::new(TokioSpawner::new()?) as Box)); let (completion_sender, completion_receiver) = std::sync::mpsc::channel(); graph_runtime.set_notifier(Arc::new(move || { let _ = completion_sender.send(()); From 8fb18012cf6a4d2039a82faf48671459eea843a4 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Sun, 2 Aug 2026 19:35:43 +0200 Subject: [PATCH 3/4] Remove unuse allow dead code --- editor/src/node_graph_executor/runtime.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/editor/src/node_graph_executor/runtime.rs b/editor/src/node_graph_executor/runtime.rs index 243bdb9e88..e3bd87086a 100644 --- a/editor/src/node_graph_executor/runtime.rs +++ b/editor/src/node_graph_executor/runtime.rs @@ -41,7 +41,6 @@ pub struct NodeRuntime { editor_preferences: EditorPreferences, old_graph: Option, update_thumbnails: bool, - #[expect(dead_code, reason = "read once the host wires a notifier onto the runtime")] graph_runtime: Arc, /// The last plain render request, replayed when an async source completion marks the graph dirty. last_render: Option, From 562e6602ebf6f9a0f6846b2f8ee89cca2902c7dc Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Mon, 3 Aug 2026 10:21:46 +0000 Subject: [PATCH 4/4] Poll source tasks once at spawn and land inline completions in the first eval --- editor/src/node_graph_executor/runtime.rs | 18 ++- node-graph/graphene-cli/src/main.rs | 12 +- .../src/dynamic_executor.rs | 4 +- .../libraries/core-types/src/runtime.rs | 121 ++++++++++++++---- node-graph/node-macro/src/codegen.rs | 34 +++-- 5 files changed, 142 insertions(+), 47 deletions(-) diff --git a/editor/src/node_graph_executor/runtime.rs b/editor/src/node_graph_executor/runtime.rs index e3bd87086a..fbfa1f09b5 100644 --- a/editor/src/node_graph_executor/runtime.rs +++ b/editor/src/node_graph_executor/runtime.rs @@ -17,7 +17,7 @@ use graphene_std::ops::ConvertAsync; use graphene_std::platform_application_io::canvas_utils::{Canvas, CanvasSurface, CanvasSurfaceHandle}; use graphene_std::raster_types::Raster; use graphene_std::renderer::{Render, RenderParams, RenderSvgSegmentList, SvgRender, SvgSegment}; -use graphene_std::runtime::{DynGraphRuntime, DynNotifier, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner}; +use graphene_std::runtime::{DynGraphRuntime, DynNotifier, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner, poll_once}; use graphene_std::transform::RenderQuality; use graphene_std::vector::Vector; use graphene_std::vector::style::RenderMode; @@ -142,8 +142,14 @@ impl Default for TokioSpawner { #[cfg(not(target_family = "wasm"))] impl Spawner for TokioSpawner { - fn spawn(&self, task: SourceFuture) { - self.0.as_ref().expect("runtime lives until drop").spawn(task); + fn spawn(&self, mut task: SourceFuture) -> bool { + let runtime = self.0.as_ref().expect("runtime lives until drop"); + let _guard = runtime.enter(); + if poll_once(&mut task) { + return true; + } + runtime.spawn(task); + false } } @@ -163,8 +169,12 @@ pub struct WasmSpawner; #[cfg(target_family = "wasm")] impl Spawner for WasmSpawner { - fn spawn(&self, task: SourceFuture) { + fn spawn(&self, mut task: SourceFuture) -> bool { + if poll_once(&mut task) { + return true; + } wasm_bindgen_futures::spawn_local(task); + false } } diff --git a/node-graph/graphene-cli/src/main.rs b/node-graph/graphene-cli/src/main.rs index 81311bdda6..a25a005c9d 100644 --- a/node-graph/graphene-cli/src/main.rs +++ b/node-graph/graphene-cli/src/main.rs @@ -13,7 +13,7 @@ use graph_craft::graphene_compiler::Compiler; use graph_craft::proto::ProtoNetwork; use graph_craft::util::load_network; use graphene_std::application_io::{ApplicationIo, NodeGraphUpdateMessage, NodeGraphUpdateSender}; -use graphene_std::runtime::{DynGraphRuntime, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner}; +use graphene_std::runtime::{DynGraphRuntime, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner, poll_once}; use interpreted_executor::dynamic_executor::DynamicExecutor; use interpreted_executor::util::wrap_network_in_scope; use std::error::Error; @@ -37,8 +37,14 @@ impl TokioSpawner { } impl Spawner for TokioSpawner { - fn spawn(&self, task: SourceFuture) { - self.0.as_ref().expect("runtime lives until drop").spawn(task); + fn spawn(&self, mut task: SourceFuture) -> bool { + let runtime = self.0.as_ref().expect("runtime lives until drop"); + let _guard = runtime.enter(); + if poll_once(&mut task) { + return true; + } + runtime.spawn(task); + false } } diff --git a/node-graph/interpreted-executor/src/dynamic_executor.rs b/node-graph/interpreted-executor/src/dynamic_executor.rs index 9aec1e605b..536e7b6a01 100644 --- a/node-graph/interpreted-executor/src/dynamic_executor.rs +++ b/node-graph/interpreted-executor/src/dynamic_executor.rs @@ -482,7 +482,9 @@ mod test { struct InertSpawner; impl Spawner for InertSpawner { - fn spawn(&self, _task: SourceFuture) {} + fn spawn(&self, _task: SourceFuture) -> bool { + false + } } #[test] diff --git a/node-graph/libraries/core-types/src/runtime.rs b/node-graph/libraries/core-types/src/runtime.rs index 45fa051721..2b64275d23 100644 --- a/node-graph/libraries/core-types/src/runtime.rs +++ b/node-graph/libraries/core-types/src/runtime.rs @@ -16,7 +16,8 @@ pub type DynRuntime = dyn Runtime + Send + Sync; pub type DynRuntime = dyn Runtime; pub trait Runtime { - fn spawn(&self, source: SourceId, future: SourceFuture); + /// Returns true when the future completed during the call, so its result is already observable. + fn spawn(&self, source: SourceId, future: SourceFuture) -> bool; } #[derive(Clone)] @@ -40,7 +41,14 @@ impl graphene_hash::CacheHash for RuntimeHandle { } pub trait Spawner { - fn spawn(&self, task: SourceFuture); + /// Returns true when the task completed during the call, so its result is already observable. + fn spawn(&self, task: SourceFuture) -> bool; +} + +/// Polls `task` once with a no-op waker, returning true if it completed. +pub fn poll_once(task: &mut SourceFuture) -> bool { + let mut context = std::task::Context::from_waker(std::task::Waker::noop()); + task.as_mut().poll(&mut context).is_ready() } #[cfg(not(target_family = "wasm"))] @@ -54,7 +62,7 @@ pub type DynNotifier = dyn Fn() + Send + Sync; pub type DynNotifier = dyn Fn(); impl Spawner for Box { - fn spawn(&self, task: SourceFuture) { + fn spawn(&self, task: SourceFuture) -> bool { (**self).spawn(task) } } @@ -63,11 +71,12 @@ impl Spawner for Box { pub struct NoopSpawner; impl Spawner for NoopSpawner { - fn spawn(&self, mut task: SourceFuture) { - let mut context = std::task::Context::from_waker(std::task::Waker::noop()); - if task.as_mut().poll(&mut context).is_pending() { - log::warn!("async source is not immediately ready and no host spawner is wired; the task is dropped"); + fn spawn(&self, mut task: SourceFuture) -> bool { + if poll_once(&mut task) { + return true; } + log::warn!("async source is not immediately ready and no host spawner is wired; the task is dropped"); + false } } @@ -132,21 +141,26 @@ impl GraphRuntime { } impl Runtime for GraphRuntime { - fn spawn(&self, source: SourceId, future: SourceFuture) { + fn spawn(&self, source: SourceId, mut future: SourceFuture) -> bool { let generations = Arc::clone(&self.generations); let dirty = Arc::clone(&self.dirty); let notifier = Arc::clone(&self.notifier); - self.spawner.spawn(Box::pin(async move { - future.await; - let mut generations = generations.lock().unwrap_or_else(PoisonError::into_inner); - if let Some(generation) = generations.get_mut(&source) { - *generation += 1; - dirty.store(true, Ordering::Release); - drop(generations); - let notifier = Arc::clone(¬ifier.lock().unwrap_or_else(PoisonError::into_inner)); - notifier(); + let mut first = true; + self.spawner.spawn(Box::pin(std::future::poll_fn(move |task_context| { + let poll = future.as_mut().poll(task_context); + if poll.is_ready() && !first { + let mut generations = generations.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(generation) = generations.get_mut(&source) { + *generation += 1; + dirty.store(true, Ordering::Release); + drop(generations); + let notifier = Arc::clone(¬ifier.lock().unwrap_or_else(PoisonError::into_inner)); + notifier(); + } } - })); + first = false; + poll + }))) } } @@ -167,8 +181,9 @@ mod tests { } impl Runtime for MockRuntime { - fn spawn(&self, source: SourceId, future: SourceFuture) { + fn spawn(&self, source: SourceId, future: SourceFuture) -> bool { self.futures.lock().unwrap().push((source, future)); + false } } @@ -192,11 +207,34 @@ mod tests { } impl Spawner for CollectSpawner { - fn spawn(&self, task: SourceFuture) { + fn spawn(&self, mut task: SourceFuture) -> bool { + if poll_once(&mut task) { + return true; + } self.tasks.lock().unwrap().push(task); + false } } + struct YieldOnce(bool); + + impl Future for YieldOnce { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, task_context: &mut std::task::Context<'_>) -> std::task::Poll<()> { + if self.0 { + return std::task::Poll::Ready(()); + } + self.0 = true; + task_context.waker().wake_by_ref(); + std::task::Poll::Pending + } + } + + fn yield_once() -> YieldOnce { + YieldOnce(false) + } + impl CollectSpawner { fn drain(&self) -> usize { let tasks = std::mem::take(&mut *self.tasks.lock().unwrap()); @@ -413,7 +451,7 @@ mod tests { let runtime = GraphRuntime::new(CollectSpawner::default()); runtime.retain_sources(&[7]); - Runtime::spawn(&runtime, 7, Box::pin(async {})); + Runtime::spawn(&runtime, 7, Box::pin(yield_once())); assert_eq!(runtime.snapshot(), vec![(7, 0)], "no bump before the future completes"); assert!(!runtime.take_dirty()); @@ -434,7 +472,7 @@ mod tests { observed.store(dirty_at_notify.load(Ordering::Acquire), Ordering::Relaxed); })); - Runtime::spawn(&runtime, 7, Box::pin(async {})); + Runtime::spawn(&runtime, 7, Box::pin(yield_once())); assert_eq!(runtime.spawner().drain(), 1); assert!(observed_dirty.load(Ordering::Relaxed), "the notifier must observe the dirty flag already set"); } @@ -447,7 +485,7 @@ mod tests { let flag = Arc::clone(¬ified); runtime.set_notifier(Arc::new(move || flag.store(true, Ordering::Relaxed))); - Runtime::spawn(&runtime, 7, Box::pin(async {})); + Runtime::spawn(&runtime, 7, Box::pin(yield_once())); runtime.retain_sources(&[]); assert_eq!(runtime.spawner().drain(), 1); assert!(!notified.load(Ordering::Relaxed)); @@ -458,7 +496,7 @@ mod tests { let runtime = GraphRuntime::new(CollectSpawner::default()); runtime.retain_sources(&[7]); - Runtime::spawn(&runtime, 7, Box::pin(async {})); + Runtime::spawn(&runtime, 7, Box::pin(yield_once())); runtime.retain_sources(&[]); assert_eq!(runtime.spawner().drain(), 1); @@ -470,7 +508,7 @@ mod tests { fn retain_sources_preserves_live_generations() { let runtime = GraphRuntime::new(CollectSpawner::default()); runtime.retain_sources(&[7]); - Runtime::spawn(&runtime, 7, Box::pin(async {})); + Runtime::spawn(&runtime, 7, Box::pin(yield_once())); runtime.spawner().drain(); runtime.retain_sources(&[7, 9]); @@ -482,9 +520,42 @@ mod tests { #[node_macro::node(category(""))] async fn epilogue_double(_: impl Ctx, value: f64) -> f64 { + yield_once().await; + value * 2. + } + + #[node_macro::node(category(""))] + async fn inline_double(_: impl Ctx, value: f64) -> f64 { value * 2. } + #[test] + fn an_immediately_ready_task_completes_inline_without_invalidating() { + let runtime = GraphRuntime::new(CollectSpawner::default()); + runtime.retain_sources(&[7]); + + assert!(Runtime::spawn(&runtime, 7, Box::pin(async {}))); + assert_eq!(runtime.spawner().drain(), 0); + assert_eq!(runtime.snapshot(), vec![(7, 0)], "inline completion must not bump the generation"); + assert!(!runtime.take_dirty()); + } + + #[test] + fn an_immediately_ready_kernel_returns_final_on_the_first_eval() { + let arena = Arena::new(64).unwrap(); + let runtime = Arc::new(GraphRuntime::new(CollectSpawner::default())); + runtime.retain_sources(&[13]); + let graph = InlineDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(13u64)); + + let snapshot = runtime.snapshot(); + let scope = EvalScope::new(None, None, None, &snapshot, &arena); + let ctx = ContextImpl::root(&scope); + assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0)); + assert!(!runtime.take_dirty()); + assert_eq!(runtime.snapshot(), vec![(13, 0)]); + assert_eq!(runtime.spawner().drain(), 0); + } + #[test] fn a_source_slot_lands_through_the_runtime_while_downstream_keys_invalidate() { let arena = Arena::new(64).unwrap(); diff --git a/node-graph/node-macro/src/codegen.rs b/node-graph/node-macro/src/codegen.rs index 85154daf93..90ecc36217 100644 --- a/node-graph/node-macro/src/codegen.rs +++ b/node-graph/node-macro/src/codegen.rs @@ -932,6 +932,22 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn }, _ => quote!(#core_types::gpoll::GPoll::Final(__future.await)), }; + let spawn_tail = |completion: TokenStream2, fallback: TokenStream2| { + quote! { + self.slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(__key, None); + let __slot = std::sync::Arc::clone(&self.slot); + if _runtime.0.spawn(_source, Box::pin(async move { + let __value = #completion; + __slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(__key, Some(__value)); + })) { + let __entries = self.slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(Some(__value)) = __entries.get(&__key) { + return __cell.merge(__value.clone()); + } + } + #fallback + } + }; let eval_tail = match (async_fn, future_kernel) { (false, false) => lift, (true, _) => { @@ -942,17 +958,12 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn .chain(data_names.iter().map(|name| quote!(self.#name.clone()))) .chain(kernel_value_names.iter().map(|name| quote!(#name.clone()))); let completion = future_completion(&parsed.output_type); + let tail = spawn_tail(completion, inflight.clone()); quote! { #slot_check - self.slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(__key, None); - let __slot = std::sync::Arc::clone(&self.slot); #(#snapshot_binding)* let __future = self::#fn_name(#(#future_args),*); - _runtime.0.spawn(_source, Box::pin(async move { - let __value = #completion; - __slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(__key, Some(__value)); - })); - #inflight + #tail } } (false, true) => { @@ -977,17 +988,12 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn _ => unreachable!("guarded by future_kernel"), }; let completion = future_completion(&payload); + let tail = spawn_tail(completion, spawn_return); quote! { #slot_check #placeholder_binding #acquire - self.slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(__key, None); - let __slot = std::sync::Arc::clone(&self.slot); - _runtime.0.spawn(_source, Box::pin(async move { - let __value = #completion; - __slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(__key, Some(__value)); - })); - #spawn_return + #tail } } };