fork/Graphite
1
0
Fork 0
mirror of https://github.com/GraphiteEditor/Graphite.git synced 2026-09-17 07:18:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Cut over to the graphene execution model

Browse Source
This commit is contained in:
Dennis Kobert
2026-07-31 13:30:21 +00:00
parent 7623b68318
commit 76ec799496
71 changed files with 3544 additions and 2378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
node-graph/libraries/application-io/Cargo.toml
View File

@@ -16,6 +16,7 @@ wgpu = ["dep:raster-types", "raster-types/wgpu"]
# Local dependencies
dyn-any = { workspace = true }
core-types = { workspace = true }
graphene-hash = { workspace = true }
vector-types = { workspace = true }
text-nodes = { workspace = true }
graphene-resource = { workspace = true }

15
node-graph/libraries/application-io/src/lib.rs
View File

@@ -21,6 +21,9 @@ pub trait ApplicationIo {
fn gpu_executor(&self) -> Option<&Self::Executor> {
None
}
fn gpu_executor_arc(&self) -> Option<Arc<Self::Executor>> {
None
}
fn load_resource(&self, hash: resource::ResourceHash) -> resource::ResourceFuture<'_>;
}
@@ -31,6 +34,10 @@ impl<T: ApplicationIo> ApplicationIo for &T {
(**self).gpu_executor()
}
fn gpu_executor_arc(&self) -> Option<Arc<T::Executor>> {
(**self).gpu_executor_arc()
}
fn load_resource(&self, hash: resource::ResourceHash) -> resource::ResourceFuture<'_> {
(**self).load_resource(hash)
}
@@ -54,7 +61,7 @@ pub trait GetEditorPreferences {
fn max_render_region_area(&self) -> u32;
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash, graphene_hash::CacheHash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum ExportFormat {
#[default]
@@ -62,14 +69,14 @@ pub enum ExportFormat {
Raster,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, DynAny)]
#[derive(Debug, Default, Clone, Copy, PartialEq, DynAny, graphene_hash::CacheHash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct TimingInformation {
pub time: f64,
pub animation_time: Duration,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, DynAny)]
#[derive(Debug, Default, Clone, Copy, PartialEq, DynAny, graphene_hash::CacheHash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct RenderConfig {
pub viewport: Footprint,
@@ -105,6 +112,7 @@ pub struct EditorApi<Io> {
pub node_graph_message_sender: Box<dyn NodeGraphUpdateSender + Send + Sync>,
/// Editor preferences made available to the graph through the `PlatformEditorApi`.
pub editor_preferences: Box<dyn GetEditorPreferences + Send + Sync>,
pub runtime: core_types::runtime::RuntimeHandle,
}
impl<Io> Eq for EditorApi<Io> {}
@@ -115,6 +123,7 @@ impl<Io: Default> Default for EditorApi<Io> {
application_io: None,
node_graph_message_sender: Box::new(Logger),
editor_preferences: Box::new(DummyPreferences),
runtime: Default::default(),
}
}
}

232
node-graph/libraries/core-types/src/context.rs
View File

@@ -496,185 +496,12 @@ impl ExtractFootprint for () {
}
}
// ==========================================
// EXTRACT TRAIT IMPLS FOR `OwnedContextImpl`
// ==========================================
// ==============
// TYPE `Context`
// ==============
impl ArcCtx for OwnedContextImpl {}
impl ExtractFootprint for OwnedContextImpl {
fn try_footprint(&self) -> Option<&Footprint> {
self.footprint.as_ref()
}
}
impl ExtractRealTime for OwnedContextImpl {
fn try_real_time(&self) -> Option<f64> {
self.real_time
}
}
impl ExtractAnimationTime for OwnedContextImpl {
fn try_animation_time(&self) -> Option<f64> {
self.animation_time
}
}
impl ExtractPointerPosition for OwnedContextImpl {
fn try_pointer_position(&self) -> Option<DVec2> {
self.pointer_position
}
}
impl ExtractPosition for OwnedContextImpl {
fn try_position(&self) -> Option<impl Iterator<Item = DVec2>> {
self.position.clone().map(|x| x.into_iter())
}
}
impl ExtractIndex for OwnedContextImpl {
fn try_index(&self) -> Option<impl Iterator<Item = usize>> {
self.index.clone().map(|x| x.into_iter())
}
}
impl ExtractVarArgs for OwnedContextImpl {
fn vararg(&self, index: usize) -> Result<DynRef<'_>, VarArgsResult> {
let Some(ref inner) = self.varargs else {
let Some(ref parent) = self.parent else {
return Err(VarArgsResult::NoVarArgs);
};
return parent.vararg(index);
};
inner.get(index).map(|x| x.as_ref() as DynRef<'_>).ok_or(VarArgsResult::IndexOutOfBounds)
}
fn varargs_len(&self) -> Result<usize, VarArgsResult> {
let Some(ref inner) = self.varargs else {
let Some(ref parent) = self.parent else {
return Err(VarArgsResult::NoVarArgs);
};
return parent.varargs_len();
};
Ok(inner.len())
}
fn hash_varargs(&self, mut hasher: &mut dyn Hasher) {
match (&self.varargs, &self.parent) {
(Some(inner), _) => {
for arg in inner.iter() {
arg.hash(&mut hasher);
}
}
(None, Some(parent)) => {
parent.hash_varargs(hasher);
}
_ => (),
};
}
}
impl CloneVarArgs for Arc<OwnedContextImpl> {
fn arc_clone(&self) -> Option<Arc<dyn ExtractVarArgs + Send + Sync>> {
Some(self.clone())
}
}
// ======================================
// TYPES `Context` AND `OwnedContextImpl`
// ======================================
pub type Context<'a> = Option<Arc<OwnedContextImpl>>;
pub type Context<'a> = ContextImpl<'a>;
type DynRef<'a> = &'a (dyn Any + Send + Sync);
type DynBox = Box<dyn AnyHash + Send + Sync>;
#[derive(dyn_any::DynAny)]
pub struct OwnedContextImpl {
parent: Option<Arc<dyn ExtractVarArgs + Sync + Send>>,
footprint: Option<Footprint>,
real_time: Option<f64>,
animation_time: Option<f64>,
pointer_position: Option<DVec2>,
position: Option<Vec<DVec2>>,
// This could be converted into a single enum to save extra bytes
index: Option<Vec<usize>>,
varargs: Option<Arc<[DynBox]>>,
}
impl std::fmt::Debug for OwnedContextImpl {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OwnedContextImpl")
.field("parent", &self.parent.as_ref().map(|_| "<Parent>"))
.field("footprint", &self.footprint)
.field("real_time", &self.real_time)
.field("animation_time", &self.animation_time)
.field("pointer_position", &self.pointer_position)
.field("index", &self.index)
.field("varargs_len", &self.varargs.as_ref().map(|x| x.len()))
.finish()
}
}
impl Default for OwnedContextImpl {
#[track_caller]
fn default() -> Self {
Self::empty()
}
}
impl graphene_hash::CacheHash for OwnedContextImpl {
fn cache_hash<H: core::hash::Hasher>(&self, state: &mut H) {
self.footprint.cache_hash(state);
self.real_time.cache_hash(state);
self.animation_time.cache_hash(state);
self.pointer_position.cache_hash(state);
self.position.cache_hash(state);
self.index.cache_hash(state);
self.hash_varargs(state);
}
}
impl OwnedContextImpl {
#[track_caller]
pub fn from<T: ExtractAll + CloneVarArgs>(value: T) -> Self {
OwnedContextImpl::from_flags(value, ContextFeatures::all())
}
#[track_caller]
pub fn from_flags<T: ExtractAll + CloneVarArgs>(value: T, bitflags: ContextFeatures) -> Self {
let parent = bitflags
.contains(ContextFeatures::VARARGS)
.then(|| match value.varargs_len() {
Ok(x) if x > 0 => value.arc_clone(),
_ => None,
})
.flatten();
let footprint = bitflags.contains(ContextFeatures::FOOTPRINT).then(|| value.try_footprint().copied()).flatten();
let real_time = bitflags.contains(ContextFeatures::REAL_TIME).then(|| value.try_real_time()).flatten();
let animation_time = bitflags.contains(ContextFeatures::ANIMATION_TIME).then(|| value.try_animation_time()).flatten();
let pointer_position = bitflags.contains(ContextFeatures::POINTER_POSITION).then(|| value.try_pointer_position()).flatten();
let position = bitflags.contains(ContextFeatures::POSITION).then(|| value.try_position()).flatten().map(|x| x.collect());
let index = bitflags.contains(ContextFeatures::INDEX).then(|| value.try_index()).flatten().map(|x| x.collect());
OwnedContextImpl {
parent,
footprint,
real_time,
animation_time,
pointer_position,
position,
index,
varargs: None,
}
}
pub const fn empty() -> Self {
OwnedContextImpl {
parent: None,
footprint: None,
real_time: None,
animation_time: None,
pointer_position: None,
position: None,
index: None,
varargs: None,
}
}
}
pub trait DynHash {
fn dyn_hash(&self, state: &mut dyn Hasher);
@@ -732,57 +559,6 @@ impl std::fmt::Debug for OwnedSlot {
}
}
impl OwnedContextImpl {
pub fn set_footprint(&mut self, footprint: Footprint) {
self.footprint = Some(footprint);
}
pub fn with_footprint(mut self, footprint: Footprint) -> Self {
self.footprint = Some(footprint);
self
}
pub fn with_real_time(mut self, real_time: f64) -> Self {
self.real_time = Some(real_time);
self
}
pub fn with_animation_time(mut self, animation_time: f64) -> Self {
self.animation_time = Some(animation_time);
self
}
pub fn with_pointer_position(mut self, pointer_position: DVec2) -> Self {
self.pointer_position = Some(pointer_position);
self
}
pub fn with_position(mut self, position: DVec2) -> Self {
if let Some(current_position) = &mut self.position {
current_position.insert(0, position);
} else {
self.position = Some(vec![position]);
}
self
}
pub fn with_index(mut self, index: usize) -> Self {
if let Some(current_index) = &mut self.index {
current_index.insert(0, index);
} else {
self.index = Some(vec![index]);
}
self
}
pub fn with_vararg(mut self, value: Box<dyn AnyHash + Send + Sync>) -> Self {
assert!(self.varargs.is_none_or(|value| value.is_empty()));
self.varargs = Some(Arc::new([value]));
self
}
pub fn into_context(self) -> Option<Arc<Self>> {
Some(Arc::new(self))
}
pub fn erase_parent(mut self) -> Self {
self.parent = None;
self
}
}
pub type SourceId = u64;
#[derive(Clone, Copy, Debug)]

17
node-graph/libraries/core-types/src/generic.rs
View File

@@ -1,17 +0,0 @@
use crate::Node;
use std::marker::PhantomData;
#[derive(Clone)]
pub struct FnNode<T: Fn(I) -> O, I, O>(T, PhantomData<(I, O)>);
impl<'i, T: Fn(I) -> O + 'i, O: 'i, I: 'i> Node<'i, I> for FnNode<T, I, O> {
type Output = O;
fn eval(&'i self, input: I) -> Self::Output {
self.0(input)
}
}
impl<T: Fn(I) -> O, I, O> FnNode<T, I, O> {
pub fn new(f: T) -> Self {
FnNode(f, PhantomData)
}
}

101
node-graph/libraries/core-types/src/lib.rs
View File

@@ -5,7 +5,6 @@ pub mod bounds;
pub mod consts;
pub mod context;
pub mod frame_table;
pub mod generic;
pub mod gpoll;
pub mod list;
pub mod math;
@@ -39,113 +38,15 @@ pub use no_std_types::blending;
pub use no_std_types::choice_type;
pub use no_std_types::color;
pub use no_std_types::shaders;
pub use node::Node;
pub use num_traits;
use std::any::TypeId;
use std::future::Future;
use std::pin::Pin;
#[cfg(feature = "wasm")]
pub use tsify;
pub use types::Cow;
// pub trait Node: for<'n> NodeIO<'n> {
/// The node trait allows for defining any node. Nodes can only take one call argument input, however they can store references to other nodes inside the struct.
/// See `node-graph/README.md` for information on how to define a new node.
pub trait Node<'i, Input> {
type Output: 'i;
/// Evaluates the node with the single specified input.
fn eval(&'i self, input: Input) -> Self::Output;
/// Resets the node, e.g. the LetNode's cache is set to None.
fn reset(&self) {}
/// Returns the name of the node for diagnostic purposes.
fn node_name(&self) -> &'static str {
std::any::type_name::<Self>()
}
/// Serialize the node which is used for the `introspect` function which can retrieve values from monitor nodes.
fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
log::warn!("Node::serialize not implemented for {}", std::any::type_name::<Self>());
None
}
}
mod types;
pub use types::*;
pub trait NodeIO<'i, Input>: Node<'i, Input>
where
Self::Output: 'i + StaticTypeSized,
Input: StaticTypeSized,
{
fn input_type(&self) -> TypeId {
TypeId::of::<Input::Static>()
}
fn input_type_name(&self) -> &'static str {
std::any::type_name::<Input>()
}
fn output_type(&self) -> TypeId {
TypeId::of::<<Self::Output as StaticTypeSized>::Static>()
}
fn output_type_name(&self) -> &'static str {
std::any::type_name::<Self::Output>()
}
fn to_node_io(&self, inputs: Vec<Type>) -> NodeIOTypes {
NodeIOTypes {
call_argument: concrete!(<Input as StaticTypeSized>::Static),
return_value: concrete!(<Self::Output as StaticTypeSized>::Static),
inputs,
}
}
fn to_async_node_io(&self, inputs: Vec<Type>) -> NodeIOTypes
where
<Self::Output as Future>::Output: StaticTypeSized,
Self::Output: Future,
{
NodeIOTypes {
call_argument: concrete!(<Input as StaticTypeSized>::Static),
return_value: future!(<<Self::Output as Future>::Output as StaticTypeSized>::Static),
inputs,
}
}
}
impl<'i, N: Node<'i, I>, I> NodeIO<'i, I> for N
where
N::Output: 'i + StaticTypeSized,
I: StaticTypeSized,
{
}
impl<'i, I: 'i, N: Node<'i, I> + ?Sized> Node<'i, I> for &'i N {
type Output = N::Output;
fn eval(&'i self, input: I) -> N::Output {
(*self).eval(input)
}
}
impl<'i, I: 'i, O: 'i, N: Node<'i, I, Output = O> + ?Sized> Node<'i, I> for Box<N> {
type Output = O;
fn eval(&'i self, input: I) -> O {
(**self).eval(input)
}
}
impl<'i, I: 'i, O: 'i, N: Node<'i, I, Output = O> + ?Sized> Node<'i, I> for std::sync::Arc<N> {
type Output = O;
fn eval(&'i self, input: I) -> O {
(**self).eval(input)
}
}
impl<'i, I, O: 'i> Node<'i, I> for Pin<Box<dyn Node<'i, I, Output = O> + 'i>> {
type Output = O;
fn eval(&'i self, input: I) -> O {
(**self).eval(input)
}
}
impl<'i, I, O: 'i> Node<'i, I> for Pin<&'i (dyn NodeIO<'i, I, Output = O> + 'i)> {
type Output = O;
fn eval(&'i self, input: I) -> O {
(**self).eval(input)
}
}
pub trait InputAccessorSource<'a, T>: InputAccessorSourceIdentifier + std::fmt::Debug {
fn get_input(&'a self, index: usize) -> Option<&'a T>;
fn set_input(&'a mut self, index: usize, value: T);

62
node-graph/libraries/core-types/src/ops.rs
View File

@@ -1,56 +1,26 @@
use crate::Node;
use crate::list::{Attribute, AttributeDyn, AttributeValueDyn, Item, List, ListDyn};
use crate::transform::Footprint;
use glam::DVec2;
use graphene_hash::CacheHash;
use std::future::Future;
use std::marker::PhantomData;
// Type
// TODO: Document this
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
pub struct TypeNode<N: for<'a> Node<'a, I>, I, O>(pub N, pub PhantomData<(I, O)>);
impl<'i, N, I: 'i, O: 'i> Node<'i, I> for TypeNode<N, I, O>
where
N: for<'n> Node<'n, I, Output = O>,
{
type Output = O;
fn eval(&'i self, input: I) -> Self::Output {
self.0.eval(input)
}
fn reset(&self) {
self.0.reset();
}
fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
self.0.serialize()
}
}
impl<'i, N: for<'a> Node<'a, I>, I: 'i> TypeNode<N, I, <N as Node<'i, I>>::Output> {
pub fn new(node: N) -> Self {
Self(node, PhantomData)
}
}
impl<'i, N: for<'a> Node<'a, I> + Clone, I: 'i> Clone for TypeNode<N, I, <N as Node<'i, I>>::Output> {
fn clone(&self) -> Self {
Self(self.0.clone(), self.1)
}
}
impl<'i, N: for<'a> Node<'a, I> + Copy, I: 'i> Copy for TypeNode<N, I, <N as Node<'i, I>>::Output> {}
/// The [`Convert`] trait allows for conversion between Rust primitive numeric types.
/// Because number casting is lossy, we cannot use the normal [`Into`] trait like we do for other types.
pub trait Convert<T, C>: Sized {
/// Converts this type into the (usually inferred) output type.
#[must_use]
fn convert(self, footprint: Footprint, converter: C) -> impl Future<Output = T> + Send;
fn convert(self, footprint: Footprint, converter: C) -> T;
}
/// The asynchronous counterpart of [`Convert`]; a conversion pair implements exactly one of the two traits.
pub trait ConvertAsync<T, C>: Sized {
#[must_use]
fn convert(self, footprint: Footprint, converter: C) -> crate::runtime::SourceFuture<T>;
}
impl<T: ToString + Send> Convert<String, ()> for T {
/// Converts this type into a `String` using its `ToString` implementation.
#[inline]
async fn convert(self, _: Footprint, _converter: ()) -> String {
fn convert(self, _: Footprint, _converter: ()) -> String {
self.to_string()
}
}
@@ -60,7 +30,7 @@ pub trait ListConvert<U> {
}
impl<U, T: ListConvert<U> + Send> Convert<List<U>, ()> for List<T> {
async fn convert(self, _: Footprint, _: ()) -> List<U> {
fn convert(self, _: Footprint, _: ()) -> List<U> {
let list: List<U> = self
.into_iter()
.map(|row| {
@@ -76,7 +46,7 @@ impl<U, T: ListConvert<U> + Send> Convert<List<U>, ()> for List<T> {
/// from any `List<U>` express their signature as `AttributeDyn` and avoid monomorphizing
/// over `U`; the compiler inserts this convert to bridge concrete-typed graph wires to the dyn input.
impl<T: Clone + Send + Sync + Default + std::fmt::Debug + PartialEq + CacheHash + 'static> Convert<AttributeDyn, ()> for List<T> {
async fn convert(self, _: Footprint, _: ()) -> AttributeDyn {
fn convert(self, _: Footprint, _: ()) -> AttributeDyn {
let values: Vec<T> = self.into_iter().map(|row| row.into_element()).collect();
AttributeDyn(Box::new(Attribute(values)))
}
@@ -86,7 +56,7 @@ impl<T: Clone + Send + Sync + Default + std::fmt::Debug + PartialEq + CacheHash
/// (such as `write_attribute`'s value-producing input) be generic over the destination list type
/// alone, with the compiler-inserted convert handling each concrete value type at the wire level.
impl<T: Clone + Send + Sync + Default + std::fmt::Debug + PartialEq + CacheHash + 'static> Convert<AttributeValueDyn, ()> for T {
async fn convert(self, _: Footprint, _: ()) -> AttributeValueDyn {
fn convert(self, _: Footprint, _: ()) -> AttributeValueDyn {
AttributeValueDyn(Box::new(self))
}
}
@@ -95,13 +65,13 @@ impl<T: Clone + Send + Sync + Default + std::fmt::Debug + PartialEq + CacheHash
/// only need attribute access (such as the `read_attribute_*` family) take a single `ListDyn` input
/// instead of monomorphizing over every possible carrier list type.
impl<T: Send> Convert<ListDyn, ()> for List<T> {
async fn convert(self, _: Footprint, _: ()) -> ListDyn {
fn convert(self, _: Footprint, _: ()) -> ListDyn {
self.into()
}
}
impl Convert<DVec2, ()> for DVec2 {
async fn convert(self, _: Footprint, _: ()) -> DVec2 {
fn convert(self, _: Footprint, _: ()) -> DVec2 {
self
}
}
@@ -115,7 +85,7 @@ pub trait FromAnchorPosition {
// Converts a position into a vector path composed of a single anchor point
impl<T: FromAnchorPosition + Send> Convert<List<T>, ()> for DVec2 {
async fn convert(self, _: Footprint, _: ()) -> List<T> {
fn convert(self, _: Footprint, _: ()) -> List<T> {
List::new_from_item(Item::new_from_element(T::from_anchor_position(self)))
}
}
@@ -124,7 +94,7 @@ impl<T: FromAnchorPosition + Send> Convert<List<T>, ()> for DVec2 {
macro_rules! impl_convert {
($from:ty, $to:ty) => {
impl Convert<$to, ()> for $from {
async fn convert(self, _: Footprint, _: ()) -> $to {
fn convert(self, _: Footprint, _: ()) -> $to {
self as $to
}
}
@@ -146,7 +116,7 @@ macro_rules! impl_convert {
impl_convert!(usize, $to);
impl Convert<DVec2, ()> for $to {
async fn convert(self, _: Footprint, _: ()) -> DVec2 {
fn convert(self, _: Footprint, _: ()) -> DVec2 {
DVec2::splat(self as f64)
}
}

636
node-graph/libraries/core-types/src/registry.rs
View File

@@ -1,10 +1,12 @@
use crate::{ContextFeature, Node, NodeIO, NodeIOTypes, ProtoNodeIdentifier, Type, WasmNotSend};
use dyn_any::{DynAny, StaticType};
use crate::concrete;
use crate::context::{Context, ContextImpl};
use crate::node::Node;
use crate::{ContextFeature, ProtoNodeIdentifier, Type, WasmNotSend, WasmNotSync};
use dyn_any::DynAny;
use graphene_hash::CacheHash;
pub use no_std_types::registry::types;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::ops::Deref;
use std::pin::Pin;
use std::hash::Hasher;
use std::sync::{LazyLock, Mutex};
// Translation struct between macro and definition
@@ -55,239 +57,457 @@ pub enum RegistryValueSource {
None,
Default(&'static str),
Scope(&'static str),
SourceId,
}
type NodeRegistry = LazyLock<Mutex<HashMap<ProtoNodeIdentifier, Vec<(NodeConstructor, NodeIOTypes)>>>>;
type NodeRegistry = LazyLock<Mutex<HashMap<ProtoNodeIdentifier, Vec<RegistryEntry>>>>;
pub static NODE_REGISTRY: NodeRegistry = LazyLock::new(|| Mutex::new(HashMap::new()));
pub static NODE_METADATA: LazyLock<Mutex<HashMap<ProtoNodeIdentifier, NodeMetadata>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
pub use crate::NodeIOTypes;
#[cfg(not(target_family = "wasm"))]
pub type DynFuture<'n, T> = Pin<Box<dyn Future<Output = T> + 'n + Send>>;
pub type ErasedNode<T> = dyn for<'c> Node<ContextImpl<'c>, Output = T> + Send + Sync;
#[cfg(target_family = "wasm")]
pub type DynFuture<'n, T> = Pin<Box<dyn std::future::Future<Output = T> + 'n>>;
pub type LocalFuture<'n, T> = Pin<Box<dyn Future<Output = T> + 'n>>;
pub type ErasedNode<T> = dyn for<'c> Node<ContextImpl<'c>, Output = T>;
#[cfg(not(target_family = "wasm"))]
pub type ErasedLendNode<T> = dyn for<'c> Node<ContextImpl<'c>, Output = &'c T> + Send + Sync;
#[cfg(target_family = "wasm")]
pub type ErasedLendNode<T> = dyn for<'c> Node<ContextImpl<'c>, Output = &'c T>;
#[cfg(not(target_family = "wasm"))]
type DynEdge = dyn std::any::Any + Send + Sync;
#[cfg(target_family = "wasm")]
type DynEdge = dyn std::any::Any;
pub fn edge_type<T: 'static>() -> Type {
Type::Fn(Box::new(concrete!(Context)), Box::new(concrete!(T)))
}
pub fn lend_edge_type<T: 'static>() -> Type {
Type::Fn(Box::new(concrete!(Context)), Box::new(Type::Ref(Box::new(concrete!(T)))))
}
pub fn cache_key<C: CacheHash + ?Sized>(ctx: &C) -> u64 {
let mut hasher = graphene_hash::FxHasher64::new();
ctx.cache_hash(&mut hasher);
hasher.finish()
}
#[derive(Debug, PartialEq)]
pub enum ConstructionError {
Arity { expected: usize, got: usize },
Type { expected: Box<Type>, found: Box<Type> },
}
pub struct SharedEdge<N: ?Sized> {
ptr: std::ptr::NonNull<N>,
own: std::sync::Arc<N>,
}
impl<N: ?Sized> SharedEdge<N> {
pub fn new(own: std::sync::Arc<N>) -> Self {
Self {
ptr: std::ptr::NonNull::from(&*own),
own,
}
}
pub fn share(&self) -> Self {
Self { ptr: self.ptr, own: self.own.clone() }
}
}
// SAFETY: `ptr` is derived from the owned Arc and never mutated through, so the edge is exactly as
// thread safe as the payload it shares.
unsafe impl<N: ?Sized + Send + Sync> Send for SharedEdge<N> {}
// SAFETY: as in Send.
unsafe impl<N: ?Sized + Send + Sync> Sync for SharedEdge<N> {}
impl<Input, N> Node<Input> for SharedEdge<N>
where
N: Node<Input> + ?Sized,
{
type Output = N::Output;
fn eval(&self, input: &Input) -> crate::gpoll::GPoll<Self::Output> {
// SAFETY: `own` keeps the payload alive for `self`'s lifetime and Arc
// payloads are address stable.
unsafe { self.ptr.as_ref() }.eval(input)
}
fn extent(&self, input: &Input) -> crate::gpoll::GPoll<crate::gpoll::Extent> {
// SAFETY: as in eval.
unsafe { self.ptr.as_ref() }.extent(input)
}
fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
// SAFETY: as in eval.
unsafe { self.ptr.as_ref() }.serialize()
}
fn eval_batch<'a>(&self, input: &'a Input, range: std::ops::Range<u64>, scratch: Option<&'a mut [std::mem::MaybeUninit<Self::Output>]>) -> crate::node::BatchStatus<'a, Self::Output>
where
Input: crate::context::InjectIndex + Copy,
{
// SAFETY: as in eval.
unsafe { self.ptr.as_ref() }.eval_batch(input, range, scratch)
}
}
pub struct EdgeHandle {
node: Box<DynEdge>,
share: fn(&DynEdge) -> Box<DynEdge>,
serialize: fn(&DynEdge) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>>,
ty: Type,
}
impl std::fmt::Debug for EdgeHandle {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("EdgeHandle").field("ty", &self.ty).finish_non_exhaustive()
}
}
// SAFETY: wasm is single threaded, so the marker-free payload never actually crosses a thread.
#[cfg(target_family = "wasm")]
unsafe impl Send for EdgeHandle {}
// SAFETY: as in Send.
#[cfg(target_family = "wasm")]
unsafe impl Sync for EdgeHandle {}
impl EdgeHandle {
pub fn new<T: 'static>(node: std::sync::Arc<ErasedNode<T>>) -> Self {
Self::new_erased(node, edge_type::<T>())
}
pub fn new_ref<T: 'static>(node: std::sync::Arc<ErasedLendNode<T>>) -> Self {
Self::new_erased(node, lend_edge_type::<T>())
}
pub fn new_erased<N>(node: std::sync::Arc<N>, ty: Type) -> Self
where
N: ?Sized + 'static + for<'c> Node<ContextImpl<'c>>,
SharedEdge<N>: WasmNotSend + WasmNotSync,
{
Self {
node: Box::new(SharedEdge::new(node)),
share: |edge| Box::new(edge.downcast_ref::<SharedEdge<N>>().expect("share hook matches the stored edge type").share()),
serialize: |edge| Node::<ContextImpl>::serialize(edge.downcast_ref::<SharedEdge<N>>().expect("serialize hook matches the stored edge type")),
ty,
}
}
pub fn ty(&self) -> &Type {
&self.ty
}
pub fn duplicate(&self) -> Self {
Self {
node: (self.share)(&*self.node),
share: self.share,
serialize: self.serialize,
ty: self.ty.clone(),
}
}
pub fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
(self.serialize)(&*self.node)
}
pub fn downcast<T: 'static>(self) -> Result<SharedEdge<ErasedNode<T>>, ConstructionError> {
self.downcast_erased(edge_type::<T>())
}
pub fn downcast_lend<T: 'static>(self) -> Result<SharedEdge<ErasedLendNode<T>>, ConstructionError> {
self.downcast_erased(lend_edge_type::<T>())
}
pub fn downcast_erased<N: ?Sized + 'static>(self, expected: Type) -> Result<SharedEdge<N>, ConstructionError> {
let found = self.ty;
self.node.downcast::<SharedEdge<N>>().map(|edge| *edge).map_err(|_| ConstructionError::Type {
expected: Box::new(expected),
found: Box::new(found),
})
}
}
pub type NodeConstructor = fn(Vec<EdgeHandle>) -> Result<EdgeHandle, ConstructionError>;
#[derive(Clone)]
pub struct RegistryEntry {
pub io: NodeIOTypes,
pub constructor: NodeConstructor,
}
pub fn construct(entry: &RegistryEntry, inputs: Vec<EdgeHandle>) -> Result<EdgeHandle, ConstructionError> {
if inputs.len() != entry.io.inputs.len() {
return Err(ConstructionError::Arity {
expected: entry.io.inputs.len(),
got: inputs.len(),
});
}
for (handle, expected) in inputs.iter().zip(&entry.io.inputs) {
if handle.ty() != expected {
return Err(ConstructionError::Type {
expected: Box::new(expected.clone()),
found: Box::new(handle.ty().clone()),
});
}
}
(entry.constructor)(inputs)
}
#[cfg(not(target_family = "wasm"))]
pub type Any<'n> = Box<dyn DynAny<'n> + 'n + Send>;
#[cfg(target_family = "wasm")]
pub type Any<'n> = Box<dyn DynAny<'n> + 'n>;
pub type FutureAny<'n> = DynFuture<'n, Any<'n>>;
// TODO: is this safe? This is assumed to be send+sync.
#[cfg(not(target_family = "wasm"))]
pub type TypeErasedNode<'n> = dyn for<'i> NodeIO<'i, Any<'i>, Output = FutureAny<'i>> + 'n + Send + Sync;
#[cfg(target_family = "wasm")]
pub type TypeErasedNode<'n> = dyn for<'i> NodeIO<'i, Any<'i>, Output = FutureAny<'i>> + 'n;
pub type TypeErasedPinnedRef<'n> = Pin<&'n TypeErasedNode<'n>>;
pub type TypeErasedRef<'n> = &'n TypeErasedNode<'n>;
pub type TypeErasedBox<'n> = Box<TypeErasedNode<'n>>;
pub type TypeErasedPinned<'n> = Pin<Box<TypeErasedNode<'n>>>;
pub type SharedNodeContainer = std::sync::Arc<NodeContainer>;
#[cfg(test)]
mod tests {
use super::*;
use crate::SourceId;
use crate::arena::Arena;
use crate::context::{Ctx, EvalScope, ExtractArena};
use crate::gpoll::GPoll;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
pub type NodeConstructor = fn(Vec<SharedNodeContainer>) -> DynFuture<'static, TypeErasedBox<'static>>;
struct CountingNode(AtomicU32);
#[derive(Clone)]
pub struct NodeContainer {
#[cfg(feature = "dealloc_nodes")]
pub node: *const TypeErasedNode<'static>,
#[cfg(not(feature = "dealloc_nodes"))]
pub node: TypeErasedRef<'static>,
}
impl<Input> Node<Input> for CountingNode {
type Output = u32;
impl Deref for NodeContainer {
type Target = TypeErasedNode<'static>;
#[cfg(feature = "dealloc_nodes")]
fn deref(&self) -> &Self::Target {
unsafe { &*(self.node) }
#[cfg(not(feature = "dealloc_nodes"))]
self.node
}
#[cfg(not(feature = "dealloc_nodes"))]
fn deref(&self) -> &Self::Target {
self.node
}
}
/// # Safety
/// Marks NodeContainer as Sync. This dissallows the use of threadlocal storage for nodes as this would invalidate references to them.
// TODO: implement this on a higher level wrapper to avoid missuse
#[cfg(feature = "dealloc_nodes")]
unsafe impl Send for NodeContainer {}
#[cfg(feature = "dealloc_nodes")]
unsafe impl Sync for NodeContainer {}
#[cfg(feature = "dealloc_nodes")]
impl Drop for NodeContainer {
fn drop(&mut self) {
unsafe { self.dealloc_unchecked() }
}
}
impl std::fmt::Debug for NodeContainer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NodeContainer").finish()
}
}
impl NodeContainer {
pub fn new(node: TypeErasedBox<'static>) -> SharedNodeContainer {
let node = Box::leak(node);
Self { node }.into()
}
#[cfg(feature = "dealloc_nodes")]
unsafe fn dealloc_unchecked(&mut self) {
unsafe {
drop(Box::from_raw(self.node as *mut TypeErasedNode));
fn eval(&self, _input: &Input) -> GPoll<u32> {
GPoll::Final(self.0.fetch_add(1, Ordering::Relaxed) + 1)
}
}
}
/// Boxes the input and downcasts the output.
/// Wraps around a node taking Box<dyn DynAny> and returning Box<dyn DynAny>
#[derive(Clone)]
pub struct DowncastBothNode<I, O> {
node: SharedNodeContainer,
_i: PhantomData<I>,
_o: PhantomData<O>,
}
impl<'input, O, I> Node<'input, I> for DowncastBothNode<I, O>
where
O: 'input + StaticType + WasmNotSend,
I: 'input + StaticType + WasmNotSend,
{
type Output = DynFuture<'input, O>;
#[inline]
#[track_caller]
fn eval(&'input self, input: I) -> Self::Output {
struct ValueNode<T>(T);
impl<T: Clone, Input> Node<Input> for ValueNode<T> {
type Output = T;
fn eval(&self, _input: &Input) -> GPoll<T> {
GPoll::Final(self.0.clone())
}
}
struct LendNode(String);
impl<'e, Input: Ctx + ExtractArena<ArenaRef = &'e Arena>> Node<Input> for LendNode {
type Output = &'e String;
fn eval(&self, input: &Input) -> GPoll<&'e String> {
match input.arena().alloc(self.0.clone()) {
Some((parked, _)) => GPoll::Final(parked),
None => GPoll::arena_exhausted(),
}
}
}
fn scope_fixture<'a>(generations: &'a [(SourceId, u64)], arena: &'a Arena) -> EvalScope<'a> {
EvalScope::new(Some(0.5), None, None, generations, arena)
}
#[test]
fn borrow_carrying_value_types_wire_through_the_general_constructor() {
struct SplitBorrow<'c>(&'c str, usize);
struct SplitNode<Node0> {
content: Node0,
}
impl<'e, Input, Node0> Node<Input> for SplitNode<Node0>
where
Input: Ctx,
Node0: Node<Input, Output = &'e String>,
{
let node_name = self.node.node_name();
let input = Box::new(input);
let future = self.node.eval(input);
Box::pin(async move {
let out = dyn_any::downcast(future.await).unwrap_or_else(|e| panic!("DowncastBothNode wrong output type: {e} in: \n{node_name}"));
*out
})
type Output = SplitBorrow<'e>;
fn eval(&self, input: &Input) -> GPoll<SplitBorrow<'e>> {
self.content.eval(input).map(|value| SplitBorrow(value, value.len()))
}
}
}
fn reset(&self) {
self.node.reset();
}
fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
self.node.serialize()
}
}
impl<I, O> DowncastBothNode<I, O> {
pub const fn new(node: SharedNodeContainer) -> Self {
Self {
node,
_i: PhantomData,
_o: PhantomData,
}
}
}
pub struct FutureWrapperNode<Node> {
node: Node,
}
type ErasedSplitEdge = dyn for<'c> Node<ContextImpl<'c>, Output = SplitBorrow<'c>> + Send + Sync;
impl<'i, T: 'i + WasmNotSend, N> Node<'i, T> for FutureWrapperNode<N>
where
N: Node<'i, T, Output: WasmNotSend> + WasmNotSend,
{
type Output = DynFuture<'i, N::Output>;
#[inline(always)]
fn eval(&'i self, input: T) -> Self::Output {
let result = self.node.eval(input);
Box::pin(async move { result })
}
#[inline(always)]
fn reset(&self) {
self.node.reset();
}
let arena = Arena::new(4096);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
#[inline(always)]
fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
self.node.serialize()
}
}
let lending = EdgeHandle::new_ref(Arc::new(LendNode("held".to_string())) as Arc<ErasedLendNode<String>>);
let upstream = lending.downcast_lend::<String>().unwrap();
let node: Arc<ErasedSplitEdge> = Arc::new(SplitNode { content: upstream });
let handle = EdgeHandle::new_erased(node, concrete!(SplitBorrow<'static>));
assert_eq!(*handle.ty(), concrete!(SplitBorrow<'static>));
impl<N> FutureWrapperNode<N> {
pub const fn new(node: N) -> Self {
Self { node }
}
}
pub struct DynAnyNode<I, O, Node> {
node: Node,
_i: PhantomData<I>,
_o: PhantomData<O>,
}
impl<'input, I, O, N> Node<'input, Any<'input>> for DynAnyNode<I, O, N>
where
I: 'input + StaticType + WasmNotSend,
O: 'input + StaticType + WasmNotSend,
N: 'input + Node<'input, I, Output = DynFuture<'input, O>>,
{
type Output = FutureAny<'input>;
#[inline]
fn eval(&'input self, input: Any<'input>) -> Self::Output {
let node_name = std::any::type_name::<N>();
let output = |input| {
let result = self.node.eval(input);
async move { Box::new(result.await) as Any<'input> }
let wired = handle.downcast_erased::<ErasedSplitEdge>(concrete!(SplitBorrow<'static>)).unwrap();
let GPoll::Final(split) = wired.eval(&ctx) else {
panic!("borrow-carrying output must eval through the erased edge");
};
match dyn_any::downcast(input) {
Ok(input) => Box::pin(output(*input)),
Err(e) => panic!("DynAnyNode Input, {e} in:\n{node_name}"),
assert_eq!(split.0, "held");
assert_eq!(split.1, 4);
}
#[test]
fn derive_ctx_repeat_pushes_index_levels_through_the_erased_edge() {
use crate::context::{DeriveCtx, Derived, ExtractIndex};
struct RepeatNode<Node0> {
content: Node0,
}
}
fn reset(&self) {
self.node.reset();
}
impl<C, T, Node0> Node<C> for RepeatNode<Node0>
where
C: Ctx + DeriveCtx,
Node0: for<'x> Node<Derived<'x, C>, Output = T>,
{
type Output = Vec<T>;
fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
self.node.serialize()
}
}
impl<'input, I, O, N> DynAnyNode<I, O, N>
where
I: 'input + StaticType,
O: 'input + StaticType,
N: 'input + Node<'input, I, Output = DynFuture<'input, O>>,
{
pub const fn new(node: N) -> Self {
Self {
node,
_i: PhantomData,
_o: PhantomData,
fn eval(&self, input: &C) -> GPoll<Vec<T>> {
let spilled = input.index_head();
let mut result = Vec::new();
for index in 0..3 {
let derived = input.promoted(&spilled, index);
match self.content.eval(&derived) {
GPoll::Final(value) => result.push(value),
other => return other.map(|_| Vec::new()),
}
}
GPoll::Final(result)
}
}
struct LevelsNode;
impl<Input: ExtractIndex> Node<Input> for LevelsNode {
type Output = Vec<usize>;
fn eval(&self, input: &Input) -> GPoll<Vec<usize>> {
GPoll::Final(input.try_index().map(|levels| levels.collect()).unwrap_or_default())
}
}
let arena = Arena::new(1024);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let nested = RepeatNode {
content: RepeatNode { content: LevelsNode },
};
let erased: Box<ErasedNode<Vec<Vec<Vec<usize>>>>> = Box::new(nested);
let GPoll::Final(outer) = erased.eval(&ctx) else {
panic!("nested repeat must evaluate");
};
assert_eq!(outer.len(), 3);
assert_eq!(outer[2][1], vec![1, 2, 0]);
assert_eq!(outer[0][0], vec![0, 0, 0]);
}
#[test]
fn derive_ctx_footprint_replace_reaches_the_content() {
use crate::context::{DeriveCtx, Derived, ExtractFootprint};
use crate::transform::Footprint;
struct ShiftFootprintNode<Node0> {
content: Node0,
}
impl<C, T, Node0> Node<C> for ShiftFootprintNode<Node0>
where
C: Ctx + DeriveCtx + ExtractFootprint,
Node0: for<'x> Node<Derived<'x, C>, Output = T>,
{
type Output = T;
fn eval(&self, input: &C) -> GPoll<T> {
let mut footprint = input.try_footprint().copied().unwrap_or(Footprint::DEFAULT);
footprint.resolution.x += 7;
let derived = input.with_footprint(&footprint);
self.content.eval(&derived)
}
}
struct ResolutionNode;
impl<Input: ExtractFootprint> Node<Input> for ResolutionNode {
type Output = u32;
fn eval(&self, input: &Input) -> GPoll<u32> {
GPoll::Final(input.try_footprint().map(|footprint| footprint.resolution.x).unwrap_or(0))
}
}
let arena = Arena::new(1024);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let graph: Box<ErasedNode<u32>> = Box::new(ShiftFootprintNode {
content: ShiftFootprintNode { content: ResolutionNode },
});
assert_eq!(graph.eval(&ctx), GPoll::Final(Footprint::DEFAULT.resolution.x + 14));
}
#[test]
fn construct_checks_arity_and_types() {
fn construct_strlen(args: Vec<EdgeHandle>) -> Result<EdgeHandle, ConstructionError> {
let mut args = args.into_iter();
let value = args.next().ok_or(ConstructionError::Arity { expected: 1, got: 0 })?.downcast::<String>()?;
drop(value);
Ok(EdgeHandle::new(Arc::new(ValueNode(0u32)) as Arc<ErasedNode<u32>>))
}
let entry = RegistryEntry {
io: NodeIOTypes::new(concrete!(Context), concrete!(u32), vec![edge_type::<String>()]),
constructor: construct_strlen,
};
let owned = EdgeHandle::new(Arc::new(ValueNode("typed".to_string())) as Arc<ErasedNode<String>>);
assert!(construct(&entry, vec![owned]).is_ok());
assert_eq!(construct(&entry, vec![]).unwrap_err(), ConstructionError::Arity { expected: 1, got: 0 });
let mistyped = EdgeHandle::new(Arc::new(ValueNode(1.0f64)) as Arc<ErasedNode<f64>>);
assert_eq!(
construct(&entry, vec![mistyped]).unwrap_err(),
ConstructionError::Type {
expected: Box::new(edge_type::<String>()),
found: Box::new(edge_type::<f64>()),
}
);
let lent = EdgeHandle::new_ref(Arc::new(LendNode("typed".to_string())) as Arc<ErasedLendNode<String>>);
assert_eq!(
construct(&entry, vec![lent]).unwrap_err(),
ConstructionError::Type {
expected: Box::new(edge_type::<String>()),
found: Box::new(lend_edge_type::<String>()),
}
);
}
#[test]
fn duplicated_edges_share_one_instance_and_outlive_each_other() {
let arena = Arena::new(1024);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let handle = EdgeHandle::new(Arc::new(CountingNode(AtomicU32::new(0))) as Arc<ErasedNode<u32>>);
let duplicate = handle.duplicate();
assert_eq!(*duplicate.ty(), edge_type::<u32>());
let first = handle.downcast::<u32>().unwrap();
let second = duplicate.downcast::<u32>().unwrap();
assert_eq!(first.eval(&ctx), GPoll::Final(1));
assert_eq!(second.eval(&ctx), GPoll::Final(2));
drop(first);
assert_eq!(second.eval(&ctx), GPoll::Final(3));
}
}
pub struct PanicNode<I: WasmNotSend, O: WasmNotSend>(PhantomData<I>, PhantomData<O>);
impl<'i, I: 'i + WasmNotSend, O: 'i + WasmNotSend> Node<'i, I> for PanicNode<I, O> {
type Output = O;
fn eval(&'i self, _: I) -> Self::Output {
unimplemented!("This node should never be evaluated")
}
}
impl<I: WasmNotSend, O: WasmNotSend> PanicNode<I, O> {
pub const fn new() -> Self {
Self(PhantomData, PhantomData)
}
}
impl<I: WasmNotSend, O: WasmNotSend> Default for PanicNode<I, O> {
fn default() -> Self {
Self::new()
}
}
// TODO: Evaluate safety
unsafe impl<I: WasmNotSend, O: WasmNotSend> Sync for PanicNode<I, O> {}

364
node-graph/libraries/core-types/src/runtime.rs
View File

@@ -146,3 +146,367 @@ impl<S: Spawner> Runtime for GraphRuntime<S> {
}));
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::arena::Arena;
use crate::context::{ContextImpl, Ctx, CtxSnapshot, EvalScope, ExtractFootprint, ExtractVarArgs, VarArgLink, VarArgSlots};
use crate::gpoll::GPoll;
use crate::node::Node;
use crate::transform::Footprint;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU32, Ordering};
#[derive(Default)]
struct MockRuntime {
futures: Mutex<Vec<(SourceId, SourceFuture)>>,
}
impl Runtime for MockRuntime {
fn spawn(&self, source: SourceId, future: SourceFuture) {
self.futures.lock().unwrap().push((source, future));
}
}
impl MockRuntime {
fn drain(&self) -> Vec<SourceId> {
let futures = std::mem::take(&mut *self.futures.lock().unwrap());
let mut task_ctx = std::task::Context::from_waker(std::task::Waker::noop());
futures
.into_iter()
.map(|(source, mut future)| {
assert!(future.as_mut().poll(&mut task_ctx).is_ready());
source
})
.collect()
}
}
#[derive(Default)]
struct CollectSpawner {
tasks: Mutex<Vec<SourceFuture>>,
}
impl Spawner for CollectSpawner {
fn spawn(&self, task: SourceFuture) {
self.tasks.lock().unwrap().push(task);
}
}
impl CollectSpawner {
fn drain(&self) -> usize {
let tasks = std::mem::take(&mut *self.tasks.lock().unwrap());
let mut task_ctx = std::task::Context::from_waker(std::task::Waker::noop());
let count = tasks.len();
for mut task in tasks {
assert!(task.as_mut().poll(&mut task_ctx).is_ready());
}
count
}
}
struct SourceNode<T>(T);
impl<T: Clone, Input> Node<Input> for SourceNode<T> {
type Output = T;
fn eval(&self, _input: &Input) -> GPoll<T> {
GPoll::Final(self.0.clone())
}
}
static SLOW_DOUBLE_RUNS: AtomicU32 = AtomicU32::new(0);
#[node_macro::node(category(""))]
async fn slow_double(_: impl Ctx, value: f64) -> f64 {
SLOW_DOUBLE_RUNS.fetch_add(1, Ordering::Relaxed);
value * 2.
}
fn stand_in(_value: &f64) -> f64 {
-1.
}
#[node_macro::node(category(""), placeholder(stand_in))]
async fn preview_double(_: impl Ctx, value: f64) -> f64 {
value * 2.
}
#[node_macro::node(category(""), placeholder(stand_in), no_partial)]
async fn strict_double(_: impl Ctx, value: f64) -> f64 {
value * 2.
}
#[node_macro::node(category(""))]
async fn snapshot_resolution(ctx: CtxSnapshot, _primary: ()) -> u32 {
ctx.try_footprint().map(|footprint| footprint.resolution.x).unwrap_or(0)
}
#[node_macro::node(category(""))]
async fn snapshot_vararg(ctx: CtxSnapshot, _primary: ()) -> f64 {
ctx.vararg(0).ok().and_then(|slot| slot.downcast_ref::<f64>()).copied().unwrap_or(0.)
}
static STAGED_RUNS: AtomicU32 = AtomicU32::new(0);
#[node_macro::node(category(""))]
fn staged_double(_: impl Ctx, value: f64) -> SourceFuture<f64> {
STAGED_RUNS.fetch_add(1, Ordering::Relaxed);
Box::pin(async move { value * 2. })
}
#[node_macro::node(category(""))]
fn staged_sum(ctx: impl Ctx, value: f64, addend: impl Node<Context<'_>, Output = f64>) -> Result<SourceFuture<f64>, crate::gpoll::Interrupt> {
let addend = addend.eval(ctx)?;
Ok(Box::pin(async move { value + addend }))
}
struct GatedSource(Arc<std::sync::atomic::AtomicBool>, f64);
impl<Input> Node<Input> for GatedSource {
type Output = f64;
fn eval(&self, _input: &Input) -> GPoll<f64> {
match self.0.load(Ordering::Relaxed) {
true => GPoll::Final(self.1),
false => GPoll::Pending,
}
}
}
fn scope_fixture<'a>(generations: &'a [(SourceId, u64)], arena: &'a Arena) -> EvalScope<'a> {
EvalScope::new(None, None, None, generations, arena)
}
#[test]
fn async_source_spawns_once_and_lands_via_the_slot() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let runtime = Arc::new(MockRuntime::default());
let graph = SlowDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(7u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
assert_eq!(SLOW_DOUBLE_RUNS.load(Ordering::Relaxed), 0);
assert_eq!(runtime.drain(), vec![7]);
assert_eq!(SLOW_DOUBLE_RUNS.load(Ordering::Relaxed), 1);
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0));
assert_eq!(SLOW_DOUBLE_RUNS.load(Ordering::Relaxed), 1);
}
#[test]
fn async_source_reports_the_placeholder_while_in_flight() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let runtime = Arc::new(MockRuntime::default());
let graph = PreviewDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(1u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Partial(-1.0));
runtime.drain();
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0));
}
#[test]
fn no_partial_maps_the_placeholder_frame_to_pending() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let runtime = Arc::new(MockRuntime::default());
let graph = StrictDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(2u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
runtime.drain();
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0));
}
#[test]
fn prologue_runs_sync_and_spawns_once() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let runtime = Arc::new(MockRuntime::default());
let graph = StagedDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(8u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
assert_eq!(STAGED_RUNS.load(Ordering::Relaxed), 1, "the prologue runs synchronously on the miss");
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
assert_eq!(STAGED_RUNS.load(Ordering::Relaxed), 1, "in flight must not rerun the prologue");
assert_eq!(runtime.drain(), vec![8]);
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0));
assert_eq!(STAGED_RUNS.load(Ordering::Relaxed), 1);
}
#[test]
fn prologue_interrupt_defers_the_spawn() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let ctx = ContextImpl::root(&scope);
let gate = Arc::new(std::sync::atomic::AtomicBool::new(false));
let runtime = Arc::new(MockRuntime::default());
let graph = StagedSumNode::new(SourceNode(40.0f64), GatedSource(gate.clone(), 2.0), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(9u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
assert_eq!(runtime.drain(), Vec::<SourceId>::new(), "an interrupted prologue must not spawn or claim the slot");
gate.store(true, Ordering::Relaxed);
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
assert_eq!(runtime.drain(), vec![9]);
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(42.0));
}
#[test]
fn async_kernels_read_captured_varargs() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let root = ContextImpl::root(&scope);
let payload = 21.5f64;
let link = VarArgLink {
args: VarArgSlots::Single(&payload),
outer: None,
};
let ctx = root.with_varargs(&link);
let runtime = Arc::new(MockRuntime::default());
let graph = SnapshotVarargNode::new(SourceNode(()), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(5u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
runtime.drain();
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(21.5));
}
#[test]
fn async_kernels_read_the_captured_context_snapshot() {
let arena = Arena::new(64);
let generations = [];
let scope = scope_fixture(&generations, &arena);
let root = ContextImpl::root(&scope);
let footprint = Footprint::DEFAULT;
let ctx = root.with_footprint(&footprint);
let runtime = Arc::new(MockRuntime::default());
let graph = SnapshotResolutionNode::new(SourceNode(()), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(3u64));
assert_eq!(Node::eval(&graph, &ctx), GPoll::Pending);
runtime.drain();
assert_eq!(Node::eval(&graph, &ctx), GPoll::Final(Footprint::DEFAULT.resolution.x));
}
#[test]
fn the_epilogue_bumps_the_generation_and_sets_dirty() {
let runtime = GraphRuntime::new(CollectSpawner::default());
runtime.retain_sources(&[7]);
Runtime::spawn(&runtime, 7, Box::pin(async {}));
assert_eq!(runtime.snapshot(), vec![(7, 0)], "no bump before the future completes");
assert!(!runtime.take_dirty());
assert_eq!(runtime.spawner().drain(), 1);
assert_eq!(runtime.snapshot(), vec![(7, 1)]);
assert!(runtime.take_dirty());
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(&notified);
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());
runtime.retain_sources(&[7]);
Runtime::spawn(&runtime, 7, Box::pin(async {}));
runtime.retain_sources(&[]);
assert_eq!(runtime.spawner().drain(), 1);
assert_eq!(runtime.snapshot(), Vec::<(SourceId, u64)>::new());
assert!(!runtime.take_dirty(), "a removed source must not invalidate");
}
#[test]
fn retain_sources_preserves_live_generations() {
let runtime = GraphRuntime::new(CollectSpawner::default());
runtime.retain_sources(&[7]);
Runtime::spawn(&runtime, 7, Box::pin(async {}));
runtime.spawner().drain();
runtime.retain_sources(&[7, 9]);
assert_eq!(runtime.snapshot(), vec![(7, 1), (9, 0)]);
runtime.retain_sources(&[9]);
assert_eq!(runtime.snapshot(), vec![(9, 0)]);
}
#[node_macro::node(category(""))]
async fn epilogue_double(_: impl Ctx, value: f64) -> f64 {
value * 2.
}
#[test]
fn a_source_slot_lands_through_the_runtime_while_downstream_keys_invalidate() {
let arena = Arena::new(64);
let runtime = Arc::new(GraphRuntime::new(CollectSpawner::default()));
runtime.retain_sources(&[11]);
let graph = EpilogueDoubleNode::new(SourceNode(21.0f64), SourceNode(RuntimeHandle(runtime.clone())), SourceNode(11u64));
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::Pending);
assert!(!runtime.take_dirty());
assert_eq!(runtime.spawner().drain(), 1);
assert!(runtime.take_dirty());
let bumped = runtime.snapshot();
assert_eq!(bumped, vec![(11, 1)]);
let bumped_scope = EvalScope::new(None, None, None, &bumped, &arena);
let bumped_ctx = ContextImpl::root(&bumped_scope);
assert_eq!(Node::eval(&graph, &bumped_ctx), GPoll::Final(42.0), "the own-generation-excluded key replays the landed slot");
assert_eq!(runtime.spawner().drain(), 0, "a slot hit must not respawn");
let downstream_key = crate::registry::cache_key(&ContextImpl::root(&scope));
let bumped_downstream_key = crate::registry::cache_key(&ContextImpl::root(&bumped_scope));
assert_ne!(downstream_key, bumped_downstream_key, "unretained keys see the bump");
}
}

201
node-graph/libraries/core-types/src/value.rs
View File

@@ -1,103 +1,18 @@
use crate::Node;
use std::cell::{Cell, RefCell, RefMut};
use std::marker::PhantomData;
#[derive(Default, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
pub struct IntNode<const N: u32>;
impl<'i, const N: u32, I> Node<'i, I> for IntNode<N> {
type Output = u32;
#[inline(always)]
fn eval(&'i self, _input: I) -> Self::Output {
N
}
}
#[derive(Default, Debug, Clone, Copy)]
pub struct ValueNode<T>(pub T);
impl<'i, T: 'i, I> Node<'i, I> for ValueNode<T> {
type Output = &'i T;
#[inline(always)]
fn eval(&'i self, _input: I) -> Self::Output {
&self.0
}
}
impl<T> ValueNode<T> {
pub const fn new(value: T) -> ValueNode<T> {
ValueNode(value)
}
}
impl<T> From<T> for ValueNode<T> {
fn from(value: T) -> Self {
ValueNode::new(value)
}
}
#[derive(Default, Debug, Clone, Copy)]
pub struct AsRefNode<T: AsRef<U>, U>(pub T, PhantomData<U>);
impl<'i, T: 'i + AsRef<U>, U: 'i> Node<'i, ()> for AsRefNode<T, U> {
type Output = &'i U;
#[inline(always)]
fn eval(&'i self, _input: ()) -> Self::Output {
self.0.as_ref()
}
}
impl<T: AsRef<U>, U> AsRefNode<T, U> {
pub const fn new(value: T) -> AsRefNode<T, U> {
AsRefNode(value, PhantomData)
}
}
#[derive(Default, Debug, Clone)]
pub struct RefCellMutNode<T>(pub RefCell<T>);
impl<'i, T: 'i> Node<'i, ()> for RefCellMutNode<T> {
type Output = RefMut<'i, T>;
#[inline(always)]
fn eval(&'i self, _input: ()) -> Self::Output {
self.0.borrow_mut()
}
}
impl<T> RefCellMutNode<T> {
pub const fn new(value: T) -> RefCellMutNode<T> {
RefCellMutNode(RefCell::new(value))
}
}
#[derive(Default)]
pub struct OnceCellNode<T>(pub Cell<T>);
impl<'i, T: Default + 'i, I> Node<'i, I> for OnceCellNode<T> {
type Output = T;
#[inline(always)]
fn eval(&'i self, _input: I) -> Self::Output {
self.0.replace(T::default())
}
}
impl<T> OnceCellNode<T> {
pub const fn new(value: T) -> OnceCellNode<T> {
OnceCellNode(Cell::new(value))
}
}
#[derive(Clone, Copy)]
pub struct ClonedNode<T: Clone>(pub T);
impl<'i, T: Clone + 'i, I> Node<'i, I> for ClonedNode<T> {
impl<T: Clone, Input> crate::node::Node<Input> for ClonedNode<T> {
type Output = T;
#[inline(always)]
fn eval(&'i self, _input: I) -> Self::Output {
self.0.clone()
fn eval(&self, _input: &Input) -> crate::gpoll::GPoll<T> {
crate::gpoll::GPoll::Final(self.0.clone())
}
}
pub fn value_edge<T: Clone + crate::WasmNotSend + crate::WasmNotSync + 'static>(value: T) -> crate::registry::EdgeHandle {
crate::registry::EdgeHandle::new(std::sync::Arc::new(ClonedNode(value)) as std::sync::Arc<crate::registry::ErasedNode<T>>)
}
impl<T: Clone> ClonedNode<T> {
pub const fn new(value: T) -> ClonedNode<T> {
ClonedNode(value)
@@ -109,103 +24,3 @@ impl<T: Clone> From<T> for ClonedNode<T> {
ClonedNode::new(value)
}
}
#[derive(Clone, Copy)]
/// The DebugClonedNode logs every time it is evaluated.
/// This is useful for debugging.
pub struct DebugClonedNode<T: Clone>(pub T);
impl<'i, T: Clone + 'i> Node<'i, ()> for DebugClonedNode<T> {
type Output = T;
#[inline(always)]
fn eval(&'i self, _input: ()) -> Self::Output {
// KEEP THIS `debug!()` - It acts as the output for the debug node itself
log::debug!("DebugClonedNode::eval");
self.0.clone()
}
}
impl<T: Clone> DebugClonedNode<T> {
pub const fn new(value: T) -> DebugClonedNode<T> {
DebugClonedNode(value)
}
}
#[derive(Clone, Copy)]
pub struct CopiedNode<T: Copy>(pub T);
impl<'i, T: Copy + 'i, I> Node<'i, I> for CopiedNode<T> {
type Output = T;
#[inline(always)]
fn eval(&'i self, _input: I) -> Self::Output {
self.0
}
}
impl<T: Copy> CopiedNode<T> {
pub const fn new(value: T) -> CopiedNode<T> {
CopiedNode(value)
}
}
#[derive(Default)]
pub struct DefaultNode<T>(PhantomData<T>);
impl<'i, T: Default + 'i, I> Node<'i, I> for DefaultNode<T> {
type Output = T;
fn eval(&'i self, _input: I) -> Self::Output {
T::default()
}
}
impl<T> DefaultNode<T> {
pub fn new() -> Self {
Self(PhantomData)
}
}
#[repr(C)]
/// Return the unit value
#[derive(Default, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
pub struct ForgetNode;
impl<'i, T: 'i> Node<'i, T> for ForgetNode {
type Output = ();
fn eval(&'i self, _input: T) -> Self::Output {}
}
impl ForgetNode {
pub const fn new() -> Self {
ForgetNode
}
}
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_int_node() {
let node = IntNode::<5>;
assert_eq!(node.eval(()), 5);
}
#[test]
fn test_value_node() {
let node = ValueNode::new(5);
assert_eq!(node.eval(()), &5);
let type_erased = &node as &dyn for<'a> Node<'a, (), Output = &'a i32>;
assert_eq!(type_erased.eval(()), &5);
}
#[test]
fn test_default_node() {
let node = DefaultNode::<u32>::new();
assert_eq!(node.eval(42), 0);
}
#[test]
#[allow(clippy::unit_cmp)]
fn test_unit_node() {
let node = ForgetNode::new();
assert_eq!(node.eval(()), ());
}
}

25
node-graph/libraries/wgpu-executor/src/lib.rs
View File

@@ -9,17 +9,16 @@ use crate::texture_cache::TextureCache;
use anyhow::Result;
use core_types::Color;
use core_types::color::SRGBA8;
use futures::lock::Mutex;
use glam::UVec2;
use graphene_application_io::{ApplicationIo, EditorApi};
use raster_types::Texture;
use std::sync::Arc;
use std::sync::Mutex;
use vello::{AaConfig, AaSupport, RenderParams, Renderer, RendererOptions, Scene};
use wgpu::{Origin3d, TextureAspect};
pub use context::Context as WgpuContext;
pub use context::ContextBuilder as WgpuContextBuilder;
pub use pipeline::AsyncPipeline as AsyncWgpuPipeline;
pub use pipeline::Pipeline as WgpuPipeline;
pub use pipeline::PipelineCache as WgpuPipelineCache;
pub use rendering::RenderContext;
@@ -61,6 +60,18 @@ impl std::fmt::Debug for WgpuExecutor {
}
}
/// Owned Arc handle carrying the executor as an ordinary wire value.
#[derive(Clone, Debug)]
pub struct WgpuExecutorHandle(pub std::sync::Arc<WgpuExecutor>);
impl std::ops::Deref for WgpuExecutorHandle {
type Target = WgpuExecutor;
fn deref(&self) -> &WgpuExecutor {
&self.0
}
}
impl<'a, T: ApplicationIo<Executor = WgpuExecutor>> From<&'a EditorApi<T>> for &'a WgpuExecutor {
fn from(editor_api: &'a EditorApi<T>) -> Self {
editor_api.application_io.as_ref().unwrap().gpu_executor().unwrap()
@@ -68,8 +79,8 @@ impl<'a, T: ApplicationIo<Executor = WgpuExecutor>> From<&'a EditorApi<T>> for &
}
impl WgpuExecutor {
pub async fn render_vello_scene(&self, scene: &Scene, size: UVec2, context: &RenderContext, background: Option<Color>) -> Result<Texture> {
let texture = self.request_texture(size).await;
pub fn render_vello_scene(&self, scene: &Scene, size: UVec2, context: &RenderContext, background: Option<Color>) -> Result<Texture> {
let texture = self.request_texture(size);
let texture_view = texture.create_view(&wgpu::TextureViewDescriptor::default());
@@ -82,7 +93,7 @@ impl WgpuExecutor {
};
{
let mut renderer = self.inner.vello_renderer.lock().await;
let mut renderer = self.inner.vello_renderer.lock().unwrap();
for (image_brush, texture) in context.resource_overrides.iter() {
let texture_view = wgpu::TexelCopyTextureInfoBase {
texture: (**texture).clone(),
@@ -109,8 +120,8 @@ impl WgpuExecutor {
pipeline.init::<P>(self);
}
pub async fn request_texture(&self, size: UVec2) -> Texture {
self.inner.texture_cache.lock().await.request_texture(&self.context().device, size)
pub fn request_texture(&self, size: UVec2) -> Texture {
self.inner.texture_cache.lock().unwrap().request_texture(&self.context().device, size)
}
}

34
node-graph/libraries/wgpu-executor/src/pipeline.rs
View File

@@ -1,42 +1,16 @@
use dyn_any::DynAny;
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, OnceLock};
use crate::WgpuExecutor;
pub type PipelineFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
pub trait Pipeline: Any + Send + Sync + Sized {
type Args<'a>;
type Out: Send;
fn create(executor: &WgpuExecutor) -> Self;
fn run<'a>(&'a self, executor: &'a WgpuExecutor, args: &'a Self::Args<'_>) -> PipelineFuture<'a, Self::Out>;
}
pub trait AsyncPipeline: Any + Send + Sync + Sized {
type Args<'a>;
type Out: Send;
fn create(executor: &WgpuExecutor) -> Self;
fn run<'a>(&'a self, executor: &'a WgpuExecutor, args: &'a Self::Args<'_>) -> impl Future<Output = Self::Out> + Send + 'a;
}
impl<P: AsyncPipeline> Pipeline for P {
type Args<'a> = <P as AsyncPipeline>::Args<'a>;
type Out = <P as AsyncPipeline>::Out;
fn create(executor: &WgpuExecutor) -> Self {
<P as AsyncPipeline>::create(executor)
}
fn run<'a>(&'a self, executor: &'a WgpuExecutor, args: &'a Self::Args<'_>) -> PipelineFuture<'a, Self::Out> {
Box::pin(<P as AsyncPipeline>::run(self, executor, args))
}
fn run<'a>(&'a self, executor: &'a WgpuExecutor, args: &'a Self::Args<'_>) -> Self::Out;
}
#[derive(Default, Clone, DynAny)]
@@ -51,13 +25,13 @@ impl PipelineCache {
self.pipeline.get_or_init(|| Box::new(P::create(executor)));
}
pub async fn run<P: Pipeline>(&self, args: &P::Args<'_>) -> P::Out {
pub fn run<P: Pipeline>(&self, args: &P::Args<'_>) -> P::Out {
let executor = self.executor.get().expect("PipelineCache not initialized");
let entry = self.pipeline.get().expect("PipelineCache not initialized");
let pipeline = (&**entry)
let pipeline = (**entry)
.downcast_ref::<P>()
.unwrap_or_else(|| panic!("PipelineCache type mismatch: run::<{}>() but init used a different pipeline type", std::any::type_name::<P>(),));
pipeline.run(executor, args).await
pipeline.run(executor, args)
}
}

6
node-graph/libraries/wgpu-executor/src/shader_runtime/per_pixel_adjust_runtime.rs
View File

@@ -2,10 +2,10 @@ use crate::WgpuContext;
use crate::shader_runtime::{FULLSCREEN_VERTEX_SHADER_NAME, ShaderRuntime};
use core_types::list::{Item, List};
use core_types::shaders::buffer_struct::BufferStruct;
use futures::lock::Mutex;
use raster_types::{GPU, Raster};
use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Mutex;
use wgpu::util::{BufferInitDescriptor, DeviceExt};
use wgpu::{
BindGroupDescriptor, BindGroupEntry, BindGroupLayoutDescriptor, BindGroupLayoutEntry, BindingResource, BindingType, Buffer, BufferBinding, BufferBindingType, BufferUsages, ColorTargetState, Face,
@@ -33,8 +33,8 @@ impl PerPixelAdjustShaderRuntime {
}
impl ShaderRuntime {
pub async fn run_per_pixel_adjust<T: BufferStruct>(&self, shaders: &Shaders<'_>, textures: List<Raster<GPU>>, args: Option<&T>) -> List<Raster<GPU>> {
let mut cache = self.per_pixel_adjust.pipeline_cache.lock().await;
pub fn run_per_pixel_adjust<T: BufferStruct>(&self, shaders: &Shaders<'_>, textures: List<Raster<GPU>>, args: Option<&T>) -> List<Raster<GPU>> {
let mut cache = self.per_pixel_adjust.pipeline_cache.lock().unwrap();
let pipeline = cache
.entry(shaders.fragment_shader_name.to_owned())
.or_insert_with(|| PerPixelAdjustGraphicsPipeline::new(&self.context, shaders));

159
node-graph/libraries/wgpu-executor/src/texture_conversion.rs
View File

@@ -1,9 +1,10 @@
use crate::WgpuExecutor;
use crate::WgpuExecutorHandle;
use core_types::Color;
use core_types::Ctx;
use core_types::color::SRGBA8;
use core_types::list::{Item, List};
use core_types::ops::Convert;
use core_types::ops::{Convert, ConvertAsync};
use core_types::runtime::SourceFuture;
use core_types::transform::Footprint;
use raster_types::Image;
use raster_types::{CPU, GPU, Raster};
@@ -38,6 +39,52 @@ fn upload_to_texture(device: &wgpu::Device, queue: &wgpu::Queue, image: &Raster<
)
}
/// Passthrough conversion for GPU `List`s - no conversion needed
impl Convert<List<Raster<GPU>>, WgpuExecutorHandle> for List<Raster<GPU>> {
fn convert(self, _: Footprint, _converter: WgpuExecutorHandle) -> List<Raster<GPU>> {
self
}
}
/// Converts a `List<Raster<CPU>>` to `List<Raster<GPU>>` by uploading each image to a texture
impl Convert<List<Raster<GPU>>, WgpuExecutorHandle> for List<Raster<CPU>> {
fn convert(self, _: Footprint, executor: WgpuExecutorHandle) -> List<Raster<GPU>> {
let device = &executor.context().device;
let queue = executor.context().queue.lock();
let list = self
.into_iter()
.map(|row| {
let (image, attributes) = row.into_parts();
let texture = upload_to_texture(device, &queue, &image);
Item::from_parts(Raster::new_gpu(texture), attributes)
})
.collect();
queue.submit([]);
list
}
}
/// Converts single CPU raster to GPU by uploading to texture
impl Convert<Raster<GPU>, WgpuExecutorHandle> for Raster<CPU> {
fn convert(self, _: Footprint, executor: WgpuExecutorHandle) -> Raster<GPU> {
let device = &executor.context().device;
let queue = executor.context().queue.lock();
let texture = upload_to_texture(device, &queue, &self);
queue.submit([]);
Raster::new_gpu(texture)
}
}
/// Passthrough conversion for CPU `List`s - no conversion needed
impl Convert<List<Raster<CPU>>, WgpuExecutorHandle> for List<Raster<CPU>> {
fn convert(self, _: Footprint, _converter: WgpuExecutorHandle) -> List<Raster<CPU>> {
self
}
}
/// Converts a Raster<GPU> texture to Raster<CPU> by downloading the underlying texture data.
///
/// Assumptions:
@@ -142,57 +189,11 @@ impl RasterGpuToRasterCpuConverter {
}
}
/// Passthrough conversion for GPU `List`s - no conversion needed
impl<'i> Convert<List<Raster<GPU>>, &'i WgpuExecutor> for List<Raster<GPU>> {
async fn convert(self, _: Footprint, _converter: &'i WgpuExecutor) -> List<Raster<GPU>> {
self
}
}
/// Converts a `List<Raster<CPU>>` to `List<Raster<GPU>>` by uploading each image to a texture
impl<'i> Convert<List<Raster<GPU>>, &'i WgpuExecutor> for List<Raster<CPU>> {
async fn convert(self, _: Footprint, executor: &'i WgpuExecutor) -> List<Raster<GPU>> {
let device = &executor.context().device;
let queue = executor.context().queue.lock();
let list = self
.into_iter()
.map(|row| {
let (image, attributes) = row.into_parts();
let texture = upload_to_texture(device, &queue, &image);
Item::from_parts(Raster::new_gpu(texture), attributes)
})
.collect();
queue.submit([]);
list
}
}
/// Converts single CPU raster to GPU by uploading to texture
impl<'i> Convert<Raster<GPU>, &'i WgpuExecutor> for Raster<CPU> {
async fn convert(self, _: Footprint, executor: &'i WgpuExecutor) -> Raster<GPU> {
let device = &executor.context().device;
let queue = executor.context().queue.lock();
let texture = upload_to_texture(device, &queue, &self);
queue.submit([]);
Raster::new_gpu(texture)
}
}
/// Passthrough conversion for CPU `List`s - no conversion needed
impl<'i> Convert<List<Raster<CPU>>, &'i WgpuExecutor> for List<Raster<CPU>> {
async fn convert(self, _: Footprint, _converter: &'i WgpuExecutor) -> List<Raster<CPU>> {
self
}
}
/// Converts a `List<Raster<GPU>>` to `List<Raster<CPU>>` by downloading texture data in one go then asynchronously maps all buffers and processes the results.
impl<'i> Convert<List<Raster<CPU>>, &'i WgpuExecutor> for List<Raster<GPU>> {
async fn convert(self, _: Footprint, executor: &'i WgpuExecutor) -> List<Raster<CPU>> {
let device = &executor.context().device;
let queue = &executor.context().queue;
impl ConvertAsync<List<Raster<CPU>>, WgpuExecutorHandle> for List<Raster<GPU>> {
fn convert(self, _: Footprint, executor: WgpuExecutorHandle) -> SourceFuture<List<Raster<CPU>>> {
let device = executor.context().device.clone();
let queue = executor.context().queue.lock();
let mut encoder = device.create_command_encoder(&wgpu::CommandEncoderDescriptor {
label: Some("batch_texture_download_encoder"),
@@ -203,48 +204,50 @@ impl<'i> Convert<List<Raster<CPU>>, &'i WgpuExecutor> for List<Raster<GPU>> {
for row in self {
let (element, attributes) = row.into_parts();
converters.push(RasterGpuToRasterCpuConverter::new(device, &mut encoder, element));
converters.push(RasterGpuToRasterCpuConverter::new(&device, &mut encoder, element));
rows_meta.push(Item::from_parts((), attributes));
}
queue.submit([encoder.finish()]);
let mut map_futures = Vec::new();
for converter in converters {
map_futures.push(converter.convert(device));
}
Box::pin(async move {
let mut map_futures = Vec::new();
for converter in converters {
map_futures.push(converter.convert(&device));
}
let map_results = futures::future::try_join_all(map_futures)
.await
.map_err(|_| "Failed to receive map result")
.expect("Buffer mapping communication failed");
let map_results = futures::future::try_join_all(map_futures)
.await
.map_err(|_| "Failed to receive map result")
.expect("Buffer mapping communication failed");
map_results
.into_iter()
.zip(rows_meta)
.map(|(element, row)| {
let (_, attributes) = row.into_parts();
Item::from_parts(element, attributes)
})
.collect()
map_results
.into_iter()
.zip(rows_meta)
.map(|(element, row)| {
let (_, attributes) = row.into_parts();
Item::from_parts(element, attributes)
})
.collect()
})
}
}
/// Converts single GPU raster to CPU by downloading texture data
impl<'i> Convert<Raster<CPU>, &'i WgpuExecutor> for Raster<GPU> {
async fn convert(self, _: Footprint, executor: &'i WgpuExecutor) -> Raster<CPU> {
let device = &executor.context().device;
let queue = &executor.context().queue;
impl ConvertAsync<Raster<CPU>, WgpuExecutorHandle> for Raster<GPU> {
fn convert(self, _: Footprint, executor: WgpuExecutorHandle) -> SourceFuture<Raster<CPU>> {
let device = executor.context().device.clone();
let queue = executor.context().queue.lock();
let mut encoder = device.create_command_encoder(&wgpu::CommandEncoderDescriptor {
label: Some("single_texture_download_encoder"),
});
let converter = RasterGpuToRasterCpuConverter::new(device, &mut encoder, self);
let converter = RasterGpuToRasterCpuConverter::new(&device, &mut encoder, self);
queue.submit([encoder.finish()]);
converter.convert(device).await.expect("Failed to download texture data")
Box::pin(async move { converter.convert(&device).await.expect("Failed to download texture data") })
}
}
@@ -252,10 +255,10 @@ impl<'i> Convert<Raster<CPU>, &'i WgpuExecutor> for Raster<GPU> {
///
/// Accepts either individual raster data or a `List` of raster elements and converts it to the GPU format using the WgpuExecutor's device and queue.
#[node_macro::node(category(""))]
pub async fn upload_texture<'a: 'n, T: Convert<List<Raster<GPU>>, &'a WgpuExecutor>>(
pub fn upload_texture<T: Convert<List<Raster<GPU>>, WgpuExecutorHandle>>(
_: impl Ctx,
#[implementations(List<Raster<CPU>>, List<Raster<GPU>>)] input: T,
executor: &'a WgpuExecutor,
executor: WgpuExecutorHandle,
) -> List<Raster<GPU>> {
input.convert(Footprint::DEFAULT, executor).await
input.convert(Footprint::DEFAULT, executor)
}