From 9e248cf90cecf657b4b76c7bc1bf3b74bbd04678 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Thu, 30 Jul 2026 19:12:29 +0000 Subject: [PATCH] Wire up async runtimes --- desktop/src/app.rs | 5 ++ desktop/wrapper/src/lib.rs | 4 + editor/src/node_graph_executor.rs | 24 +++++- editor/src/node_graph_executor/runtime.rs | 79 +++++++++++++++++-- node-graph/graphene-cli/src/export.rs | 36 +++++++-- node-graph/graphene-cli/src/main.rs | 35 +++++++- .../src/dynamic_executor.rs | 13 +-- .../libraries/core-types/src/runtime.rs | 45 +++++++++++ 8 files changed, 213 insertions(+), 28 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 94642736ab..2cbc23297a 100644 --- a/editor/src/node_graph_executor.rs +++ b/editor/src/node_graph_executor.rs @@ -26,7 +26,7 @@ pub use runtime_io::NodeRuntimeIO; mod runtime; pub use runtime::*; -#[derive(Debug, serde::Serialize, serde::Deserialize)] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct ExecutionRequest { execution_id: u64, render_config: RenderConfig, @@ -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(), @@ -375,10 +379,22 @@ impl NodeGraphExecutor { } } - let Some((queued_execution_id, execution_context)) = self.futures.pop_front() else { - panic!("InvalidGenerationId") + 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 { + // 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"), + } }; - assert_eq!(queued_execution_id, execution_id, "Missmatch in execution id"); // TODO: Eventually remove this document upgrade code // Gradient-migration measurement runs only read back the fill's evaluated geometry; they never render to the artwork. diff --git a/editor/src/node_graph_executor/runtime.rs b/editor/src/node_graph_executor/runtime.rs index 36f7b87d70..c35d667a4e 100644 --- a/editor/src/node_graph_executor/runtime.rs +++ b/editor/src/node_graph_executor/runtime.rs @@ -16,7 +16,8 @@ use graphene_std::ops::{Convert, 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::core_types::gpoll::GPoll; +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; @@ -40,6 +41,9 @@ pub struct NodeRuntime { editor_preferences: EditorPreferences, old_graph: Option, update_thumbnails: bool, + 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, @@ -119,9 +123,51 @@ 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 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 graph_runtime: Arc = Arc::new(GraphRuntime::new(Box::new(NoopSpawner) as Box)); + #[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)); @@ -133,6 +179,8 @@ impl NodeRuntime { old_graph: None, resources: ResourceRegistry::default(), update_thumbnails: true, + graph_runtime: Arc::clone(&graph_runtime), + last_render: None, editor_api: PlatformEditorApi { editor_preferences: Box::new(EditorPreferences::default()), @@ -178,6 +226,9 @@ impl NodeRuntime { } let for_export = execution_request.render_config.for_export; + if !for_export { + self.last_render = Some(execution_request.clone()); + } execution = Some(request); @@ -198,6 +249,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 { @@ -383,11 +438,16 @@ impl NodeRuntime { async fn execute_network(&mut self, render_config: RenderConfig) -> Result { use graph_craft::graphene_compiler::Executor; - let (value, evaluation_error) = (&self.executor).execute(render_config).await.map_err(|e| e.to_string())?; - if let Some(error) = evaluation_error { - error!("Node graph evaluation reported an error alongside its fallback output: {error:?}"); + match (&self.executor).execute(render_config).await.map_err(|e| e.to_string())? { + GPoll::Final(value) | GPoll::Partial(value) => Ok(value), + GPoll::Fallback(boxed) => { + let (value, error) = *boxed; + error!("Node graph evaluation reported an error alongside its fallback output: {error:?}"); + Ok(value) + } + GPoll::Pending => Err("Node graph evaluation is pending".to_string()), + GPoll::Error(error) => Err(format!("Node graph evaluation failed: {error:?}")), } - Ok(value) } /// Updates state data @@ -576,6 +636,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 a80f745e19..7f76022769 100644 --- a/node-graph/graphene-cli/src/export.rs +++ b/node-graph/graphene-cli/src/export.rs @@ -1,6 +1,7 @@ use graph_craft::document::value::{RenderOutputType, TaggedValue, UVec2}; use graph_craft::graphene_compiler::Executor; use graphene_std::application_io::{ExportFormat, RenderConfig, TimingInformation}; +use graphene_std::core_types::gpoll::GPoll; use graphene_std::core_types::ops::ConvertAsync; use graphene_std::core_types::transform::Footprint; use graphene_std::raster_types::{CPU, GPU, Raster}; @@ -8,8 +9,31 @@ 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; +const SOURCE_COMPLETION_TIMEOUT: Duration = Duration::from_secs(30); + +async fn execute_until_final(executor: &DynamicExecutor, render_config: RenderConfig, completion: &Receiver<()>) -> Result> { + loop { + while completion.try_recv().is_ok() {} + match executor.execute(render_config.clone()).await? { + GPoll::Final(value) => return Ok(value), + GPoll::Fallback(boxed) => { + let (value, error) = *boxed; + log::error!("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()), + } + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum FileType { Svg, @@ -36,6 +60,7 @@ pub async 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 { @@ -57,10 +82,7 @@ pub async fn export_document( } // Execute the graph - let (result, evaluation_error) = executor.execute(render_config).await?; - if let Some(error) = evaluation_error { - log::error!("Node graph evaluation reported an error alongside its fallback output: {error:?}"); - } + let result = execute_until_final(executor, render_config, completion).await?; // Handle the result based on output type match result { @@ -159,6 +181,7 @@ pub async 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}; @@ -198,10 +221,7 @@ pub async fn export_gif( } // Execute the graph for this frame - let (result, evaluation_error) = executor.execute(render_config).await?; - if let Some(error) = evaluation_error { - log::error!("Node graph evaluation reported an error alongside its fallback output: {error:?}"); - } + let result = execute_until_final(executor, render_config, completion).await?; // 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 24d036123b..a4d9562461 100644 --- a/node-graph/graphene-cli/src/main.rs +++ b/node-graph/graphene-cli/src/main.rs @@ -14,7 +14,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; @@ -29,6 +29,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 { @@ -183,7 +206,11 @@ async 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 {}), @@ -230,9 +257,9 @@ async 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).await?; + export::export_gif(&executor, wgpu_executor_ref.clone(), output, scale, (width, height), animation, &completion_receiver).await?; } else { - export::export_document(&executor, wgpu_executor_ref.clone(), output, file_type, scale, (width, height), transparent).await?; + export::export_document(&executor, wgpu_executor_ref.clone(), output, file_type, scale, (width, height), transparent, &completion_receiver).await?; } } _ => unreachable!("All other commands should be handled before this match statement is run"), diff --git a/node-graph/interpreted-executor/src/dynamic_executor.rs b/node-graph/interpreted-executor/src/dynamic_executor.rs index 589820f482..74ea205699 100644 --- a/node-graph/interpreted-executor/src/dynamic_executor.rs +++ b/node-graph/interpreted-executor/src/dynamic_executor.rs @@ -170,11 +170,11 @@ impl DynamicExecutor { } } -impl Executor)> for &DynamicExecutor +impl Executor> for &DynamicExecutor where I: VarArg + Send + Sync + std::panic::RefUnwindSafe, { - fn execute(&self, input: I) -> LocalFuture<'_, Result<(TaggedValue, Option), Box>> { + fn execute(&self, input: I) -> LocalFuture<'_, Result, Box>> { Box::pin(async move { let Some(handle) = self.tree.get(self.output) else { return Err("Output node not found in executor".into()); @@ -185,13 +185,14 @@ where Err(error) => GPoll::Final(Err(error)), }); match result { - GPoll::Final(value) | GPoll::Partial(value) => Ok((value?, None)), + GPoll::Final(value) => Ok(GPoll::Final(value?)), + GPoll::Partial(value) => Ok(GPoll::Partial(value?)), GPoll::Fallback(boxed) => { let (value, error) = *boxed; - Ok((value?, Some(error))) + Ok(GPoll::Fallback(Box::new((value?, error)))) } - GPoll::Pending => Err("Node graph evaluation is pending".into()), - GPoll::Error(error) => Err(format!("Node graph evaluation failed: {error:?}").into()), + GPoll::Pending => Ok(GPoll::Pending), + GPoll::Error(error) => Ok(GPoll::Error(error)), } }) } diff --git a/node-graph/libraries/core-types/src/runtime.rs b/node-graph/libraries/core-types/src/runtime.rs index b0163fa2b3..164f5bfe80 100644 --- a/node-graph/libraries/core-types/src/runtime.rs +++ b/node-graph/libraries/core-types/src/runtime.rs @@ -48,6 +48,11 @@ pub type DynSpawner = dyn Spawner + Send + Sync; #[cfg(target_family = "wasm")] pub type DynSpawner = dyn Spawner; +#[cfg(not(target_family = "wasm"))] +pub type DynNotifier = dyn Fn() + Send + Sync; +#[cfg(target_family = "wasm")] +pub type DynNotifier = dyn Fn(); + impl Spawner for Box { fn spawn(&self, task: SourceFuture) { (**self).spawn(task) @@ -74,6 +79,7 @@ impl Default for RuntimeHandle { pub struct GraphRuntime { generations: Arc>>, dirty: Arc, + notifier: Arc>>, spawner: S, } @@ -89,10 +95,15 @@ impl GraphRuntime { Self { generations: Arc::default(), dirty: Arc::default(), + notifier: Arc::new(Mutex::new(Arc::new(|| {}))), spawner, } } + pub fn set_notifier(&self, notifier: Arc) { + *self.notifier.lock().unwrap_or_else(PoisonError::into_inner) = notifier; + } + pub fn retain_sources(&self, live: &[SourceId]) { let mut generations = self.generations.lock().unwrap_or_else(PoisonError::into_inner); generations.retain(|source, _| live.contains(source)); @@ -121,12 +132,16 @@ impl Runtime for GraphRuntime { fn spawn(&self, source: SourceId, future: SourceFuture) { 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(); } })); } @@ -405,6 +420,36 @@ mod tests { assert!(!runtime.take_dirty(), "take_dirty drains the flag"); } + #[test] + fn the_epilogue_notifies_after_setting_dirty() { + let runtime = GraphRuntime::new(CollectSpawner::default()); + runtime.retain_sources(&[7]); + let observed_dirty = Arc::new(AtomicBool::new(false)); + let dirty_at_notify = Arc::clone(&runtime.dirty); + let observed = Arc::clone(&observed_dirty); + runtime.set_notifier(Arc::new(move || { + observed.store(dirty_at_notify.load(Ordering::Acquire), Ordering::Relaxed); + })); + + Runtime::spawn(&runtime, 7, Box::pin(async {})); + assert_eq!(runtime.spawner().drain(), 1); + assert!(observed_dirty.load(Ordering::Relaxed), "the notifier must observe the dirty flag already set"); + } + + #[test] + fn the_epilogue_of_a_removed_source_does_not_notify() { + let runtime = GraphRuntime::new(CollectSpawner::default()); + runtime.retain_sources(&[7]); + let notified = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(¬ified); + runtime.set_notifier(Arc::new(move || flag.store(true, Ordering::Relaxed))); + + Runtime::spawn(&runtime, 7, Box::pin(async {})); + runtime.retain_sources(&[]); + assert_eq!(runtime.spawner().drain(), 1); + assert!(!notified.load(Ordering::Relaxed)); + } + #[test] fn the_epilogue_of_a_removed_source_is_inert() { let runtime = GraphRuntime::new(CollectSpawner::default());