Add graphene-cli host, join and watch

Author: TrueDoctor

Cargo.lock

@@ -18,6 +18,7 @@ members = [
 	"document/migrations/snapshot",
 	"document/migrations/nodes",
 	"document/peer-transport",
+	"document/live",
 	"editor",
 	"frontend/wrapper",
 	"libraries/dyn-any",
@@ -115,6 +116,7 @@ migration-runner = { path = "document/migrations/runner" }
 migration-formats = { path = "document/migrations/formats" }
 migration-nodes = { path = "document/migrations/nodes" }
 peer-transport = { path = "document/peer-transport" }
+document-live = { path = "document/live" }
 raster-nodes = { path = "node-graph/nodes/raster" }
 graphene-std = { path = "node-graph/nodes/gstd" }
 interpreted-executor = { path = "node-graph/interpreted-executor" }

Cargo.toml

@@ -137,6 +137,12 @@ impl<L: Layout + Send + Sync> LoadResource for Gdd<L> {
 
 pub struct ResourceProxy<T: Layout>(AnyContainer, T);
 
+impl<T: Layout + Clone> Clone for ResourceProxy<T> {
+	fn clone(&self) -> Self {
+		Self(self.0.clone(), self.1.clone())
+	}
+}
+
 impl<L: Layout + Send + Sync> LoadResource for ResourceProxy<L> {
 	fn load(&self, hash: ResourceHash) -> ResourceFuture<'_> {
 		Box::pin(async move {

document/format/src/resource.rs

@@ -0,0 +1,21 @@
+[package]
+name = "document-live"
+description = "A memory-backed document following or hosting a collaborative session, for viewers and embedders"
+version.workspace = true
+license.workspace = true
+authors.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+publish.workspace = true
+
+[dependencies]
+document-container = { workspace = true }
+document-format = { workspace = true, features = ["network"] }
+document-graph-storage = { workspace = true }
+peer-transport = { workspace = true }
+graph-craft = { workspace = true }
+thiserror = { workspace = true }
+log = { workspace = true }
+
+[lints]
+workspace = true

document/live/Cargo.toml

@@ -0,0 +1,124 @@
+//! A memory-backed document hosting or following a collaborative session, handing out its runtime `NodeNetwork`
+//! on demand. Owns no executor or async runtime, so the caller spawns the transport driver.
+
+use document_container::AnyContainer;
+use document_container::backends::memory::MemoryBackend;
+use document_format::{GddV1, GddV1Layout, ResourceProxy};
+use document_graph_storage::{PeerId, UserId};
+use graph_craft::application_io::resource::ResourceHash;
+use graph_craft::document::NodeNetwork;
+use peer_transport::{Room, SyncTarget};
+
+pub use document_format::Error as FormatError;
+pub use document_graph_storage::to_runtime::ConversionError;
+pub use peer_transport::{DEFAULT_SIGNALING_SERVER, Event, MessageLoopFuture, Role, SessionToken};
+
+#[derive(Debug, thiserror::Error)]
+pub enum LiveError {
+	#[error(transparent)]
+	Format(#[from] FormatError),
+	#[error(transparent)]
+	Conversion(#[from] ConversionError),
+}
+
+pub struct LiveDocument {
+	gdd: GddV1,
+	token: SessionToken,
+}
+
+impl LiveDocument {
+	/// Create an empty document and host a session under `token`. `spawn` must drive the returned
+	/// future for as long as the session lives.
+	pub async fn host(signaling_server: &str, token: SessionToken, peer: PeerId, user: UserId, spawn: impl FnOnce(MessageLoopFuture)) -> Result<Self, LiveError> {
+		let gdd = Self::empty_document(peer, user).await?;
+		Self::host_document(gdd, signaling_server, token, user, spawn)
+	}
+
+	pub fn host_document(mut gdd: GddV1, signaling_server: &str, token: SessionToken, user: UserId, spawn: impl FnOnce(MessageLoopFuture)) -> Result<Self, LiveError> {
+		gdd.share(Self::connect(signaling_server, token, spawn), user)?;
+		Ok(Self { gdd, token })
+	}
+
+	/// Join the session under `token`. The document fills in once the host's sync arrives.
+	pub async fn join(signaling_server: &str, token: SessionToken, peer: PeerId, user: UserId, spawn: impl FnOnce(MessageLoopFuture)) -> Result<Self, LiveError> {
+		let mut gdd = Self::empty_document(peer, user).await?;
+		gdd.connect(Self::connect(signaling_server, token, spawn), user)?;
+		Ok(Self { gdd, token })
+	}
+
+	fn connect(signaling_server: &str, token: SessionToken, spawn: impl FnOnce(MessageLoopFuture)) -> Room {
+		let (room, driver) = Room::connect(&token.signaling_url(signaling_server));
+		spawn(driver);
+		room
+	}
+
+	async fn empty_document(peer: PeerId, user: UserId) -> Result<GddV1, FormatError> {
+		let version = env!("CARGO_PKG_VERSION").to_string();
+		GddV1::create_in(AnyContainer::Memory(MemoryBackend::new()), GddV1Layout, peer, user, peer.0, version.clone(), version)
+	}
+
+	pub fn token(&self) -> SessionToken {
+		self.token
+	}
+
+	pub fn role(&self) -> Option<Role> {
+		self.gdd.role()
+	}
+
+	pub fn is_synced(&self) -> bool {
+		self.gdd.is_synced()
+	}
+
+	/// Synced and holding every resource the document references, so [`network`](Self::network) can be built and run.
+	pub fn is_ready(&self) -> bool {
+		self.is_synced() && SyncTarget::missing_resources(&self.gdd).is_empty()
+	}
+
+	/// Referenced resources whose bytes no peer has sent yet.
+	pub fn missing_resources(&self) -> Vec<ResourceHash> {
+		let mut missing: Vec<ResourceHash> = SyncTarget::missing_resources(&self.gdd).into_iter().collect();
+		missing.sort_unstable();
+		missing
+	}
+
+	/// Apply what peers sent since the last call and answer their resource requests. Call it regularly.
+	pub async fn poll(&mut self) -> Vec<Event> {
+		let events = self.gdd.poll_peers();
+		for event in &events {
+			let Event::ResourceRequested { from, hash } = event else { continue };
+			match self.gdd.read_resource(hash).await {
+				Ok(bytes) => {
+					if let Err(error) = self.gdd.send_resource(*from, *hash, bytes.as_slice().to_vec()) {
+						log::warn!("Failed to send resource {hash}: {error}");
+					}
+				}
+				Err(error) => log::warn!("Peer asked for resource {hash} which is not in this document: {error}"),
+			}
+		}
+		events
+	}
+
+	pub fn document(&self) -> &GddV1 {
+		&self.gdd
+	}
+
+	pub fn document_mut(&mut self) -> &mut GddV1 {
+		&mut self.gdd
+	}
+
+	/// The runtime graph, rebuilt from the registry. Async because proto-node declarations are resources.
+	pub async fn network(&self) -> Result<NodeNetwork, LiveError> {
+		let declarations = self.gdd.declarations(&self.gdd).await;
+		let (network, _metadata) = self.gdd.registry().to_runtime_with_metadata(&declarations)?;
+		Ok(network)
+	}
+
+	/// Byte store of resources received from peers, for an executor to load from.
+	pub fn resource_proxy(&self) -> ResourceProxy<GddV1Layout> {
+		self.gdd.resource_proxy()
+	}
+
+	pub fn leave(&mut self) {
+		self.gdd.leave();
+	}
+}

document/live/src/lib.rs

@@ -29,16 +29,20 @@ document-format = { workspace = true, features = ["zip", "xz", "migrations"] }
 document-container = { workspace = true, features = ["zip", "xz"] }
 migration-formats = { workspace = true }
 migration-nodes = { workspace = true }
+document-graph-storage = { workspace = true }
+document-live = { workspace = true }
 
 # Workspace dependencies
 log = { workspace = true }
 fern = { workspace = true }
 chrono = { workspace = true }
 wgpu = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["io-util", "time"] }
+winit = { workspace = true }
 clap = { workspace = true, features = ["cargo", "derive"] }
 image = { workspace = true }
 serde_json = { workspace = true }
+getrandom = "0.3"
 wgpu-executor = { workspace = true, optional = true }
 
 [package.metadata.cargo-shear]

node-graph/graphene-cli/Cargo.toml

@@ -1,7 +1,7 @@
 use crate::TokioSpawner;
 use graph_craft::document::value::{DAffine2, DVec2, RenderOutputType, TaggedValue, UVec2};
 use graph_craft::graphene_compiler::Executor;
-use graphene_std::application_io::{ExportFormat, RenderConfig, TimingInformation};
+use graphene_std::application_io::{ExportFormat, RenderConfig, Texture, TimingInformation};
 use graphene_std::core_types::ops::ConvertAsync;
 use graphene_std::core_types::transform::Footprint;
 use graphene_std::raster_types::{CPU, GPU, Raster};
@@ -206,6 +206,17 @@ pub(crate) async fn render_frame(
 	}
 }
 
+/// Execute the graph for one frame, keeping the result on the GPU.
+pub async fn render_texture(executor: &DynamicExecutor, render_config: RenderConfig, runtime: &AsyncRuntime, spawner: &TokioSpawner) -> Result<Texture, Box<dyn Error>> {
+	match execute_until_settled(executor, runtime, spawner, render_config).await? {
+		TaggedValue::RenderOutput(output) => match output.data {
+			RenderOutputType::Texture(texture) => Ok(texture),
+			other => Err(format!("Expected a GPU render output, got: {other:?}").into()),
+		},
+		other => Err(format!("Expected RenderOutput, got: {other:?}").into()),
+	}
+}
+
 /// Writes a rendered frame to a file of the export's type.
 fn write_frame(path: &Path, settings: ExportSettings, frame: RenderedFrame) -> Result<(), Box<dyn Error>> {
 	let contents = match frame {

node-graph/graphene-cli/src/export.rs

@@ -7,13 +7,14 @@ pub mod export;
 use document_container::AnyContainer;
 use document_container::backends::memory::MemoryBackend;
 use document_format::{GddV1, GddV1Layout};
+use graph_craft::application_io::resource::LoadResource;
 use graph_craft::application_io::{PlatformApplicationIo, PlatformEditorApi};
 use graph_craft::document::{NetworkLocation, NodeId, NodeNetwork, OwnedNetworkLocation};
 use graph_craft::graphene_compiler::Compiler;
 use graph_craft::proto::ProtoNetwork;
 use graph_craft::util::load_network;
 use graphene_std::application_io::ApplicationIo;
-use graphene_std::runtime::{AsyncRuntime, Spawn, Task};
+use graphene_std::runtime::{AsyncRuntime, RuntimeHandle, Spawn, Task};
 use interpreted_executor::dynamic_executor::DynamicExecutor;
 use interpreted_executor::util::wrap_network_in_scope;
 use std::collections::HashMap;
@@ -93,12 +94,12 @@ pub struct CompiledDocument {
 	spawner: Arc<TokioSpawner>,
 }
 
-/// Compile `document` against a fresh GPU context, resolving its resources from its `.gdd`.
-pub async fn compile_document(document: OpenedDocument) -> Result<CompiledDocument, Box<dyn Error>> {
+/// A GPU context and the editor API handed to compiled graphs, resolving resources through `resource_proxy` when given.
+pub async fn editor_api(resource_proxy: Option<Box<dyn LoadResource>>) -> (Arc<PlatformApplicationIo>, Arc<PlatformEditorApi>) {
 	log::info!("Creating GPU context");
 	let mut application_io = PlatformApplicationIo::new().await;
-	if let Some(gdd) = &document.gdd {
-		application_io.inject_resource_proxy(Box::new(gdd.resource_proxy()));
+	if let Some(resource_proxy) = resource_proxy {
+		application_io.inject_resource_proxy(resource_proxy);
 	}
 	let application_io = Arc::new(application_io);
 	if let Some(executor) = application_io.gpu_executor() {
@@ -109,14 +110,15 @@ pub async fn compile_document(document: OpenedDocument) -> Result<CompiledDocume
 	let editor_api = Arc::new(PlatformEditorApi {
 		application_io: Some(application_io.clone()),
 	});
-	let spawner = Arc::new(TokioSpawner::default());
-	let runtime = AsyncRuntime::new(spawner.clone(), Arc::new(std::time::Instant::now()));
+	(application_io, editor_api)
+}
 
-	let mut network = wrap_network_in_scope(document.network, editor_api, runtime.handle());
+/// Compile `network`. A `.gdd` resolves resource hashes from its registry, while a legacy `.graphite` has no resource store.
+pub fn compile_graph(network: NodeNetwork, editor_api: Arc<PlatformEditorApi>, gdd: Option<&GddV1>, runtime: RuntimeHandle) -> Result<ProtoNetwork, Box<dyn Error>> {
+	let mut network = wrap_network_in_scope(network, editor_api, runtime);
 
-	// A `.gdd` resolves resource hashes from its registry, while a legacy `.graphite` has no resource store
 	let preprocessor = preprocessor::Preprocessor::new();
-	let preprocessed = match &document.gdd {
+	let preprocessed = match gdd {
 		Some(gdd) => preprocessor.preprocess(&mut network, &|resource_id| gdd.registry().resources.get(&resource_id).and_then(|resource| resource.hash)),
 		None => preprocessor.preprocess(&mut network, &|_| None),
 	};
@@ -126,9 +128,26 @@ pub async fn compile_document(document: OpenedDocument) -> Result<CompiledDocume
 	for warning in &compiled_network.warnings {
 		eprintln!("{warning:?}");
 	}
+	Ok(compiled_network.network)
+}
+
+/// An executor for `proto_network`, with its errors joined into one.
+pub fn create_executor(proto_network: ProtoNetwork) -> Result<DynamicExecutor, Box<dyn Error>> {
+	DynamicExecutor::new(proto_network).map_err(|errors| {
+		let messages: Vec<String> = errors.iter().map(|error| format!("{error:?}")).collect();
+		messages.join("\n").into()
+	})
+}
+
+/// Compile `document` against a fresh GPU context, resolving its resources from its `.gdd`.
+pub async fn compile_document(document: OpenedDocument) -> Result<CompiledDocument, Box<dyn Error>> {
+	let (application_io, editor_api) = editor_api(document.gdd.as_ref().map(|gdd| Box::new(gdd.resource_proxy()) as Box<dyn LoadResource>)).await;
+	let spawner = Arc::new(TokioSpawner::default());
+	let runtime = AsyncRuntime::new(spawner.clone(), Arc::new(std::time::Instant::now()));
+	let proto_network = compile_graph(document.network, editor_api, document.gdd.as_ref(), runtime.handle())?;
 
 	Ok(CompiledDocument {
-		proto_network: compiled_network.network,
+		proto_network,
 		application_io,
 		runtime,
 		spawner,
@@ -137,10 +156,7 @@ pub async fn compile_document(document: OpenedDocument) -> Result<CompiledDocume
 
 impl CompiledDocument {
 	pub fn into_renderer(self) -> Result<Renderer, Box<dyn Error>> {
-		let executor = DynamicExecutor::new(self.proto_network).map_err(|errors| {
-			let messages: Vec<String> = errors.iter().map(|error| format!("{error:?}")).collect();
-			messages.join("\n")
-		})?;
+		let executor = create_executor(self.proto_network)?;
 		self.runtime.retain_sources(executor.sources());
 
 		let device = self.application_io.gpu_executor().ok_or("No GPU executor is available")?.context().device.clone();