Implement async refactor (#4731)

* Add editor task module, drop NodeGraphUpdateSender

* Remove Node::reset and unused value nodes

* Remove unneeded async from vector nodes

* Remove unneeded async from graphic nodes

* Remove unneeded async from raster nodes

* Remove unneeded async from the remaining nodes

* Make most node fns private

* Make node construction sync

* Remove block_on around node construction

* Move GPU readback into ConvertAsync

* Make Convert sync

* Make wgpu pipelines and vello rendering sync

* Make convert, brush and render nodes sync

* Pass editor API as owned Arc

* Take scope inputs by value

* Move Spawn and Task into core-types

* Add Poll type and status tracking

* Add source generations to context

* Stop storing context features in the document

* Track source ids in context modification

* Add async runtime and memo table

* Pass runtime through scope, expose source ids

* Set up async runtime in editor, CLI and benches

* Give async nodes runtime and source id inputs

* Classify node return types in the macro

* Make Node eval sync and return Poll

* Make lazy input nodes, tests and benches sync

* Rerender when a spawned task finishes

* Rerender in CLI until all tasks are done

* Abort pending tasks when node is dropped

* Trace errors through Poll node inputs

* Let node errors carry upstream errors

* Resolve error traces to the origin node

* Show evaluation errors in the node graph

* Default unmapped item primaries to a value

* Add debug nodes

* Memoize fallback results too

* Show multiple node graph errors

* Report node errors in CLI

* Attach input fallback error as upstream

* Keep input fallback error when node fails

* Keep Poll status in render cache

* Keep last frame while graph is pending

* Treat partial export output as pending

* Move environment from global to dispatcher

* Add editor builder, drop global node runtime

* Add Wake message, rerun graph when tasks finish

* Add source generation tests

* Add test for node error origin

* Add render status overlay

* Replace error artwork with render status

* Poll node graph through portfolio message

* Minor review fixups

* Remove future type and macro

---------

Co-authored-by: Dennis Kobert <dennis@kobert.dev>

Author: timon-schelling

Cargo.lock

@@ -22,10 +22,11 @@ use crate::input::InputState;
 use crate::persist;
 use crate::preferences;
 use crate::render::{RenderError, RenderState};
+use crate::spawner::TokioSpawner;
 use crate::ui::{InputEvent, UiCommand, UiInstance};
 use crate::window::Window;
 use crate::wrapper::messages::{DesktopFrontendMessage, DesktopWrapperMessage, IngestAction, Preferences};
-use crate::wrapper::{DesktopWrapper, FolderStore, MmapResourceStorage, NodeGraphExecutionResult, WgpuContext, serialize_frontend_messages};
+use crate::wrapper::{DesktopWrapper, FolderStore, MmapResourceStorage, WgpuContext, serialize_frontend_messages};
 
 pub(crate) struct App {
 	render_state: Option<RenderState>,
@@ -77,36 +78,47 @@ impl App {
 
 		let exiting = Arc::new(AtomicBool::new(false));
 
-		let rendering_app_event_scheduler = app_event_scheduler.clone();
 		let (start_render_sender, start_render_receiver) = std::sync::mpsc::sync_channel(1);
-		let exiting_clone = exiting.clone();
-		std::thread::spawn(move || {
-			let runtime = tokio::runtime::Runtime::new().unwrap();
-			loop {
-				let result = runtime.block_on(DesktopWrapper::execute_node_graph());
-				rendering_app_event_scheduler.schedule(AppEvent::NodeGraphExecutionResult(result));
-				let _ = start_render_receiver.recv_timeout(Duration::from_millis(10));
-				if exiting_clone.load(Ordering::Relaxed) {
-					break;
-				}
-			}
-		});
+
+		let async_runtime = tokio::runtime::Builder::new_multi_thread()
+			.thread_name("graphite-async")
+			.enable_all()
+			.build()
+			.expect("failed to construct tokio runtime");
 
 		let resource_storage = MmapResourceStorage::new(dirs::app_resources_dir()).expect("Failed to initialize on-disk resource storage");
 
-		// Wake the winit event loop when an editor future completes.
 		let wake_scheduler = app_event_scheduler.clone();
-		let wake = Arc::new(move || {
+		let editor_wake = move || {
 			wake_scheduler.schedule(AppEvent::DesktopWrapperMessage(DesktopWrapperMessage::Wake));
-		});
-		let desktop_wrapper = DesktopWrapper::new(
+		};
+		let runtime_wake_sender = start_render_sender.clone();
+		let runtime_wake = move || {
+			let _ = runtime_wake_sender.try_send(());
+		};
+		let (desktop_wrapper, mut node_runtime) = DesktopWrapper::new(
 			rand::rng().random(),
 			Arc::new(resource_storage),
 			Arc::new(FolderStore::new(dirs::app_autosave_documents_dir())),
 			wgpu_context.clone(),
-			wake,
+			TokioSpawner::new(async_runtime.handle().clone()),
+			editor_wake,
+			runtime_wake,
 		);
 
+		let rendering_app_event_scheduler = app_event_scheduler.clone();
+		let exiting_clone = exiting.clone();
+		std::thread::spawn(move || {
+			loop {
+				let texture = async_runtime.block_on(node_runtime.run());
+				rendering_app_event_scheduler.schedule(AppEvent::ViewportTexture(texture));
+				let _ = start_render_receiver.recv_timeout(Duration::from_millis(10));
+				if exiting_clone.load(Ordering::Relaxed) {
+					break;
+				}
+			}
+		});
+
 		Self {
 			render_state: None,
 			wgpu_context,
@@ -436,19 +448,15 @@ impl App {
 				}
 			}
 			AppEvent::DesktopWrapperMessage(message) => self.dispatch_desktop_wrapper_message(message),
-			AppEvent::NodeGraphExecutionResult(result) => match result {
-				NodeGraphExecutionResult::HasRun(texture) => {
-					self.dispatch_desktop_wrapper_message(DesktopWrapperMessage::PollNodeGraphEvaluation);
-					if let Some(texture) = texture
-						&& let Some(render_state) = self.render_state.as_mut()
-						&& let Some(window) = self.window.as_ref()
-					{
-						render_state.bind_viewport_texture(texture);
-						window.request_redraw();
-					}
+			AppEvent::ViewportTexture(texture) => {
+				if let Some(texture) = texture
+					&& let Some(render_state) = self.render_state.as_mut()
+					&& let Some(window) = self.window.as_ref()
+				{
+					render_state.bind_viewport_texture(texture);
+					window.request_redraw();
 				}
-				NodeGraphExecutionResult::NotRun => {}
-			},
+			}
 			AppEvent::UiUpdate(texture) => {
 				if let Some(render_state) = self.render_state.as_mut() {
 					render_state.bind_ui_texture(texture);

desktop/src/app.rs

@@ -1,13 +1,13 @@
 use crate::ui::Cursor;
-use crate::wrapper::NodeGraphExecutionResult;
+use crate::wrapper::Texture;
 use crate::wrapper::messages::DesktopWrapperMessage;
 
 pub(crate) enum AppEvent {
 	UiUpdate(wgpu::Texture),
 	CursorChange(Cursor),
 	WebCommunicationInitialized,
 	DesktopWrapperMessage(DesktopWrapperMessage),
-	NodeGraphExecutionResult(NodeGraphExecutionResult),
+	ViewportTexture(Option<Texture>),
 	ClipboardRead,
 	ClipboardWrite {
 		content: String,

desktop/src/event.rs

@@ -22,6 +22,7 @@ mod persist;
 mod preferences;
 mod render;
 mod socket;
+mod spawner;
 mod window;
 
 pub(crate) mod consts;

desktop/src/lib.rs

@@ -228,7 +228,7 @@ impl RenderState {
 			return;
 		};
 		let size = glam::UVec2::new(viewport_texture.width(), viewport_texture.height());
-		let result = futures::executor::block_on(self.executor.render_vello_scene(&scene, size, &Default::default(), None));
+		let result = self.executor.render_vello_scene(&scene, size, &Default::default(), None);
 		match result {
 			Ok(texture) => {
 				self.overlays_texture = Some(texture);

desktop/src/render/state.rs

@@ -0,0 +1,13 @@
+use crate::wrapper::{Spawn, Task};
+
+pub(crate) struct TokioSpawner(tokio::runtime::Handle);
+impl TokioSpawner {
+	pub(crate) fn new(handle: tokio::runtime::Handle) -> Self {
+		Self(handle)
+	}
+}
+impl Spawn for TokioSpawner {
+	fn spawn(&self, task: Task) {
+		self.0.spawn(task);
+	}
+}

desktop/src/spawner.rs

@@ -10,7 +10,7 @@ pub(super) fn handle_desktop_wrapper_message(dispatcher: &mut DesktopWrapperMess
 			dispatcher.queue_editor_message(*message);
 		}
 		DesktopWrapperMessage::Wake => {
-			dispatcher.queue_editor_message(EditorMessage::Future(FutureMessage::Wake));
+			dispatcher.queue_editor_message(EditorMessage::Wake);
 		}
 		DesktopWrapperMessage::Input(message) => {
 			dispatcher.queue_editor_message(EditorMessage::InputPreprocessor(message));
@@ -35,7 +35,6 @@ pub(super) fn handle_desktop_wrapper_message(dispatcher: &mut DesktopWrapperMess
 				dispatcher.respond(DesktopFrontendMessage::WriteFile { path, content });
 			}
 		},
-		DesktopWrapperMessage::PollNodeGraphEvaluation => dispatcher.poll_node_graph_evaluation(),
 		DesktopWrapperMessage::UpdateMaximized { maximized } => {
 			let message = FrontendMessage::UpdateMaximized { maximized };
 			dispatcher.queue_editor_message(message);

desktop/wrapper/src/handle_desktop_wrapper_message.rs

@@ -1,8 +1,9 @@
 use graph_craft::application_io::PlatformApplicationIo;
 use graph_craft::application_io::resource::ResourceStorage;
 use graphite_editor::application::{Editor, Environment, Host, Platform};
+use graphite_editor::builder::{Builder, EditorBuilder, RuntimeBuilder};
 use graphite_editor::messages::frontend::FrontendMessage;
-use graphite_editor::messages::prelude::Wake;
+use graphite_editor::node_graph_executor::NodeRuntime;
 
 use message_dispatcher::DesktopWrapperMessageDispatcher;
 use messages::{DesktopFrontendMessage, DesktopWrapperMessage};
@@ -11,6 +12,7 @@ use std::sync::Arc;
 pub use graph_craft::application_io::resource::MmapResourceStorage;
 pub use graphite_editor::consts::{DOUBLE_CLICK_MILLISECONDS, FILE_EXTENSION};
 pub use graphite_editor::document_store::{DocumentStore, FolderStore};
+pub use graphite_editor::task::{Spawn, Task};
 pub use wgpu_executor::Texture;
 pub use wgpu_executor::WgpuBackends;
 pub use wgpu_executor::WgpuContext;
@@ -33,7 +35,15 @@ pub struct DesktopWrapper {
 }
 
 impl DesktopWrapper {
-	pub fn new(uuid_random_seed: u64, resource_storage: Arc<dyn ResourceStorage>, document_store: Arc<dyn DocumentStore>, wgpu_context: WgpuContext, schedule_wake: Wake) -> Self {
+	pub fn new(
+		random_seed: u64,
+		resource_storage: Arc<dyn ResourceStorage>,
+		document_store: Arc<dyn DocumentStore>,
+		wgpu_context: WgpuContext,
+		spawner: impl Spawn + Send + Sync + 'static,
+		editor_wake: impl Fn() + Send + Sync + 'static,
+		runtime_wake: impl Fn() + Send + Sync + 'static,
+	) -> (Self, NodeRuntime) {
 		#[cfg(target_os = "windows")]
 		let host = Host::Windows;
 		#[cfg(target_os = "macos")]
@@ -43,29 +53,27 @@ impl DesktopWrapper {
 		let env = Environment { platform: Platform::Desktop, host };
 		let application_io = PlatformApplicationIo::new_with_context(wgpu_context);
 
-		Self {
-			editor: Editor::new(env, uuid_random_seed, resource_storage, document_store, application_io, schedule_wake),
-		}
+		let (editor, runtime) = Builder::new()
+			.with_editor(
+				EditorBuilder::new()
+					.with_wake(editor_wake)
+					.with_environment(env)
+					.with_resource_storage(resource_storage)
+					.with_document_store(document_store),
+			)
+			.with_runtime(RuntimeBuilder::new().with_wake(runtime_wake).with_application_io(application_io))
+			.with_spawner(spawner)
+			.with_random_seed(random_seed)
+			.build();
+
+		(Self { editor }, runtime)
 	}
 
 	pub fn dispatch(&mut self, message: DesktopWrapperMessage) -> Vec<DesktopFrontendMessage> {
 		let mut executor = DesktopWrapperMessageDispatcher::new(&mut self.editor);
 		executor.queue_desktop_wrapper_message(message);
 		executor.execute()
 	}
-
-	pub async fn execute_node_graph() -> NodeGraphExecutionResult {
-		let result = graphite_editor::node_graph_executor::run_node_graph().await;
-		match result {
-			(true, texture) => NodeGraphExecutionResult::HasRun(texture),
-			(false, _) => NodeGraphExecutionResult::NotRun,
-		}
-	}
-}
-
-pub enum NodeGraphExecutionResult {
-	HasRun(Option<Texture>),
-	NotRun,
 }
 
 pub fn deserialize_editor_message(data: &[u8]) -> Option<DesktopWrapperMessage> {

desktop/wrapper/src/lib.rs

@@ -42,18 +42,6 @@ impl<'a> DesktopWrapperMessageDispatcher<'a> {
 		self.responses.push(response);
 	}
 
-	pub(super) fn poll_node_graph_evaluation(&mut self) {
-		let mut responses = VecDeque::new();
-		if let Err(e) = self.editor.poll_node_graph_evaluation(&mut responses)
-			&& e != "No active document"
-		{
-			tracing::error!("Error polling node graph: {}", e);
-		}
-		while let Some(message) = responses.pop_front() {
-			self.queue_editor_message(message);
-		}
-	}
-
 	fn process_queue(&mut self) {
 		let mut frontend_messages = Vec::new();
 

desktop/wrapper/src/message_dispatcher.rs

@@ -79,7 +79,6 @@ pub enum DesktopWrapperMessage {
 	Input(InputMessage),
 	IngestFile { path: PathBuf, content: Vec<u8>, action: IngestAction },
 	SaveFileDialogResult { path: PathBuf, context: SaveFileDialogContext },
-	PollNodeGraphEvaluation,
 	UpdateMaximized { maximized: bool },
 	UpdateFullscreen { fullscreen: bool },
 	LoadPersistedState { state: PersistedState },