fix(nodes/plan-screen): video output is not applied to plan screen or…

… ndi outputs

crates/components/connections/protocols/dmx/src/processor.rs

@@ -6,6 +6,13 @@ use crate::DmxConnectionManager;
 pub(crate) struct DmxProcessor;
 
 impl Processor for DmxProcessor {
+    fn priorities(&self) -> ProcessorPriorities {
+        ProcessorPriorities {
+            post_process: 500,
+            ..Default::default()
+        }
+    }
+
     #[tracing::instrument]
     fn post_process(&mut self, injector: &mut Injector, _: ClockFrame) {
         profiling::scope!("DmxProcessor::post_process");

crates/components/fixtures/src/processor.rs

@@ -13,6 +13,14 @@ use crate::{FixtureId, FixtureState};
 pub struct FixtureProcessor;
 
 impl Processor for FixtureProcessor {
+    fn priorities(&self) -> ProcessorPriorities {
+        ProcessorPriorities {
+            pre_process: -100,
+            post_process: 100,
+            ..Default::default()
+        }
+    }
+
     #[tracing::instrument]
     fn pre_process(&mut self, injector: &mut Injector, _: ClockFrame, _fps: f64) {
         profiling::scope!("FixtureProcessor::pre_process");

crates/runtime/processing/src/lib.rs

@@ -4,7 +4,18 @@ pub use mizer_clock::ClockFrame;
 pub use mizer_debug_ui::{DebugUi, DebugUiDrawHandle, DebugUiPane, NodeStateAccess};
 pub use mizer_injector::{Inject, Injector};
 
+#[derive(Default, Debug, Clone, Copy)]
+pub struct ProcessorPriorities {
+    pub pre_process: i32,
+    pub process: i32,
+    pub post_process: i32,
+}
+
 pub trait Processor {
+    fn priorities(&self) -> ProcessorPriorities {
+        ProcessorPriorities::default()
+    }
+
     fn pre_process(&mut self, _injector: &mut Injector, _frame: ClockFrame, _fps: f64) {}
     fn process(&mut self, _injector: &mut Injector, _frame: ClockFrame) {}
     fn post_process(&mut self, _injector: &mut Injector, _frame: ClockFrame) {}

crates/runtime/src/coordinator.rs

@@ -283,15 +283,27 @@ impl Runtime for CoordinatorRuntime {
         }
         self.clock_snapshot.set(snapshot);
         tracing::trace!("pre_process");
-        for processor in self.processors.iter_mut() {
+        for processor in self
+            .processors
+            .iter_mut()
+            .sorted_by_key(|processor| processor.priorities().pre_process)
+        {
             processor.pre_process(&mut self.injector, frame, fps);
         }
         tracing::trace!("process");
-        for processor in self.processors.iter_mut() {
+        for processor in self
+            .processors
+            .iter_mut()
+            .sorted_by_key(|processor| processor.priorities().process)
+        {
             processor.process(&mut self.injector, frame);
         }
         tracing::trace!("post_process");
-        for processor in self.processors.iter_mut() {
+        for processor in self
+            .processors
+            .iter_mut()
+            .sorted_by_key(|processor| processor.priorities().post_process)
+        {
             processor.post_process(&mut self.injector, frame);
         }
         if self

crates/runtime/src/pipeline.rs

@@ -11,7 +11,7 @@ use mizer_node::{
 use mizer_nodes::{ContainerNode, Node, NodeDowncast, NodeExt};
 use mizer_pipeline::{NodePortReader, NodePreviewRef, PipelineWorker, ProcessingNodeExt};
 use mizer_ports::PortId;
-use mizer_processing::Processor;
+use mizer_processing::{Processor, ProcessorPriorities};
 use mizer_project_files::{Channel, Project};
 use pinboard::NonEmptyPinboard;
 use std::cmp::Ordering;
@@ -586,6 +586,14 @@ fn boxed_node(node: Node) -> Box<dyn ProcessingNodeExt> {
 pub struct RuntimeProcessor;
 
 impl Processor for RuntimeProcessor {
+    fn priorities(&self) -> ProcessorPriorities {
+        ProcessorPriorities {
+            pre_process: -50,
+            process: 0,
+            post_process: 50,
+        }
+    }
+
     fn pre_process(&mut self, injector: &mut Injector, frame: ClockFrame, fps: f64) {
         profiling::scope!("Pipeline::pre_process");
         let (pipeline, injector) = injector.get_slice_mut::<Pipeline>().unwrap();

crates/runtime/wgpu/src/pipeline.rs

@@ -13,7 +13,11 @@ const U32_SIZE: u32 = std::mem::size_of::<u32>() as u32;
 pub struct WgpuPipeline {
     command_buffers: Arc<Mutex<Vec<wgpu::CommandBuffer>>>,
     surfaces: Arc<Mutex<Vec<WindowSurface>>>,
+    // List of currently acquired transfer buffers
     transfer_buffers: Arc<Mutex<WeakKeyHashMap<Weak<BufferHandle>, wgpu::Buffer>>>,
+    // List of buffers that are queued to be mapped this frame
+    queued_buffers: Arc<Mutex<WeakHashSet<Weak<BufferHandle>>>>,
+    // List of buffers that were actually mapped this frame
     mapped_buffers: Arc<Mutex<WeakHashSet<Weak<BufferHandle>>>>,
 }
 
@@ -117,7 +121,7 @@ impl WgpuPipeline {
 
             self.add_stage(encoder.finish());
             tracing::trace!("Queuing buffer mapping {buffer_handle:?}");
-            self.mapped_buffers.lock().insert(buffer_handle);
+            self.queued_buffers.lock().insert(buffer_handle);
         } else {
             tracing::warn!("Unknown buffer handle");
         }
@@ -126,14 +130,18 @@ impl WgpuPipeline {
     pub(crate) fn map_buffers(&self, wgpu_context: &WgpuContext) {
         profiling::scope!("WgpuPipeline::map_buffers");
         let transfer_buffers = self.transfer_buffers.lock();
-        let mapped_buffers = self.mapped_buffers.lock();
-        for handle in mapped_buffers.iter() {
+        let queued_buffers = self.queued_buffers.lock();
+        for handle in queued_buffers.iter() {
             if let Some(buffer) = transfer_buffers.get(&handle) {
                 tracing::trace!("Mapping buffer {handle:?}");
                 let buffer_slice = buffer.slice(..);
+                let mapped_buffers = self.mapped_buffers.clone();
                 buffer_slice.map_async(wgpu::MapMode::Read, move |result| {
                     if let Err(err) = result {
                         tracing::error!("Failed to map buffer: {err:?}");
+                    } else {
+                        tracing::trace!("Buffer {handle:?} mapped successfully");
+                        mapped_buffers.lock().insert(handle);
                     }
                 });
             }
@@ -158,18 +166,20 @@ impl WgpuPipeline {
     pub(crate) fn cleanup(&self) {
         self.transfer_buffers.lock().remove_expired();
         self.unmap_buffers();
+        self.queued_buffers.lock().clear();
+        self.mapped_buffers.lock().clear();
     }
 
+    // Unmap buffer so it can be mapped in the next frame again
     pub(crate) fn unmap_buffers(&self) {
         let transfer_buffers = self.transfer_buffers.lock();
-        let mut mapped_buffers = self.mapped_buffers.lock();
+        let mapped_buffers = self.mapped_buffers.lock();
         for handle in mapped_buffers.iter() {
             if let Some(buffer) = transfer_buffers.get(&handle) {
                 tracing::trace!("Unmapping buffer {handle:?}");
                 buffer.unmap();
             }
         }
-        mapped_buffers.clear();
     }
 }
 
@@ -183,6 +193,7 @@ impl BufferAccess<'_> {
     pub fn read_mut(&self) -> Option<wgpu::BufferViewMut> {
         profiling::scope!("BufferAccess::read_mut");
         if !self.mapped_buffers.contains(&self.handle) {
+            tracing::debug!("Buffer not mapped yet");
             return None;
         }
         let buffer = self.transfer_buffers.get(&self.handle)?;
@@ -195,6 +206,7 @@ impl BufferAccess<'_> {
     pub fn read(&self) -> Option<wgpu::BufferView> {
         profiling::scope!("BufferAccess::read");
         if !self.mapped_buffers.contains(&self.handle) {
+            tracing::debug!("Buffer not mapped yet");
             return None;
         }
         let buffer = self.transfer_buffers.get(&self.handle)?;

crates/runtime/wgpu/src/processor.rs

@@ -1,10 +1,18 @@
-use mizer_module::{ClockFrame, Injector, Processor};
+use mizer_module::{ClockFrame, Inject, Injector, Processor, ProcessorPriorities};
 
 use crate::{WgpuContext, WgpuPipeline};
 
 pub struct WgpuPipelineProcessor;
 
 impl Processor for WgpuPipelineProcessor {
+    fn priorities(&self) -> ProcessorPriorities {
+        ProcessorPriorities {
+            process: 100,
+            post_process: 100,
+            ..Default::default()
+        }
+    }
+
     fn process(&mut self, injector: &mut Injector, _frame: ClockFrame) {
         profiling::scope!("WgpuPipelineProcessor::process");
         let pipeline = injector.get::<WgpuPipeline>().unwrap();
@@ -16,7 +24,7 @@ impl Processor for WgpuPipelineProcessor {
 
     fn post_process(&mut self, injector: &mut Injector, _frame: ClockFrame) {
         profiling::scope!("WgpuPipelineProcessor::post_process");
-        let pipeline = injector.get::<WgpuPipeline>().unwrap();
+        let pipeline = injector.inject::<WgpuPipeline>();
 
         pipeline.cleanup();
     }