From 086eabafbfd544fe094c7cacb846d2f6dea3053f Mon Sep 17 00:00:00 2001 From: Per Enstedt Date: Fri, 2 Oct 2026 13:23:14 +0200 Subject: [PATCH] test(vision_mixer): register the cudadownload stand-in once, never remove it vision_mixer_cuda_input_test crashed in about one run in four (11 of 40 locally, and on CI in PR #949) with an ElementFactory instance_of assertion or a segfault. Each test registered a stand-in cudadownload and removed it from the registry afterwards, so the missing-adapter test could see it absent. Registry::remove_feature frees the factory while the element class keeps a plain pointer to it, so the next registration or lookup under that name read freed memory; the lock between the tests did not help. The stand-in is now registered once per process and never removed, and the missing-adapter test moves to its own binary, vision_mixer_cuda_missing_test, where nothing is registered. The shared Media Player -> Vision Mixer harness moves to tests/vision_mixer_cuda/. The test bodies are unchanged. 40 of 40 runs of each binary pass. Co-Authored-By: Claude Opus 5.5 --- backend/tests/vision_mixer_cuda/mod.rs | 188 ++++++++++++ backend/tests/vision_mixer_cuda_input_test.rs | 287 ++---------------- .../tests/vision_mixer_cuda_missing_test.rs | 55 ++++ 3 files changed, 272 insertions(+), 258 deletions(-) create mode 100644 backend/tests/vision_mixer_cuda/mod.rs create mode 100644 backend/tests/vision_mixer_cuda_missing_test.rs diff --git a/backend/tests/vision_mixer_cuda/mod.rs b/backend/tests/vision_mixer_cuda/mod.rs new file mode 100644 index 000000000..d0e759770 --- /dev/null +++ b/backend/tests/vision_mixer_cuda/mod.rs @@ -0,0 +1,188 @@ +//! The Media Player -> GPU Vision Mixer harness shared by the CUDA input tests. +//! +//! Only the Media Player's elements are started: the mixer stays in NULL, so no +//! GL context is needed. The queries that decide the input front happen while +//! the Media Player's `identity` settles its caps, before any CAPS event or +//! buffer reaches the mixer. + +use gstreamer as gst; +use gstreamer::prelude::*; +use std::collections::HashMap; +use std::time::{Duration, Instant}; +use strom::blocks::BlockRegistry; +use strom::events::EventBroadcaster; +use strom::gst::pipeline::PipelineManager; +use strom_types::{Flow, PropertyValue as PV}; +use tempfile::NamedTempFile; + +/// The GL elements the GPU Vision Mixer is built from. Their absence is a CI +/// regression, not a reason to skip: a skip would pass green guarding nothing. +pub const GL_ELEMENTS: &[&str] = &["glupload", "glcolorconvert", "glvideomixerelement"]; + +pub const MIXER: &str = "vm"; +pub const PLAYER: &str = "player"; + +pub const CUDA_CAPS: &str = "video/x-raw(memory:CUDAMemory),format=NV12,width=320,height=240,\ + framerate=25/1,pixel-aspect-ratio=1/1,interlace-mode=progressive"; +pub const SYSTEM_CAPS: &str = "video/x-raw,format=NV12,width=320,height=240,\ + framerate=25/1,pixel-aspect-ratio=1/1,interlace-mode=progressive"; + +pub fn block( + id: &str, + definition: &str, + properties: Vec<(&str, PV)>, +) -> strom_types::BlockInstance { + let properties: HashMap = properties + .into_iter() + .map(|(k, v)| (k.to_string(), v)) + .collect(); + let computed_external_pads = strom::blocks::builtin::get_builder(definition) + .and_then(|builder| builder.get_external_pads(&properties)); + strom_types::BlockInstance { + id: id.to_string(), + block_definition_id: definition.to_string(), + name: None, + properties, + position: strom_types::block::Position { x: 0.0, y: 0.0 }, + runtime_data: None, + computed_external_pads, + } +} + +/// A Media Player (decode mode, empty playlist) feeding `mixer_input` of a GPU +/// Vision Mixer. +pub fn player_into_mixer(name: &str, mixer_input: &str) -> Flow { + let mut flow = Flow::new(name.to_string()); + flow.blocks.push(block( + MIXER, + "builtin.vision_mixer", + vec![ + ("compositor_preference", PV::String("gpu".to_string())), + ("num_inputs", PV::UInt(2)), + ("num_dsk_inputs", PV::String("1".to_string())), + ("pgm_resolution", PV::String("320x240".to_string())), + ("multiview_resolution", PV::String("320x240".to_string())), + ], + )); + flow.blocks.push(block( + PLAYER, + "builtin.media_player", + vec![("decode", PV::Bool(true))], + )); + flow.links.push(strom_types::Link { + from: format!("{}:video_out", PLAYER), + to: format!("{}:{}", MIXER, mixer_input), + }); + flow +} + +pub fn build_manager(flow: &Flow) -> (PipelineManager, NamedTempFile) { + gst::init().unwrap(); + strom::gpu::detect_gpu_capabilities(); + let temp_file = NamedTempFile::new().unwrap(); + let registry = BlockRegistry::new(temp_file.path()); + let manager = PipelineManager::new( + flow, + EventBroadcaster::with_capacity(10), + ®istry, + vec![], + "all".to_string(), + None, + std::env::temp_dir(), + std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())), + ) + .expect("GPU vision mixer flow should build"); + (manager, temp_file) +} + +pub fn element(pipeline: &gst::Pipeline, name: &str) -> gst::Element { + pipeline + .by_name(name) + .unwrap_or_else(|| panic!("{} exists", name)) +} + +/// The element feeding `name`'s sink pad. +pub fn feeder_of(pipeline: &gst::Pipeline, name: &str) -> Option { + element(pipeline, name) + .static_pad("sink")? + .peer()? + .parent_element() +} + +pub fn factory_name(element: &gst::Element) -> String { + element + .factory() + .map(|f| f.name().to_string()) + .unwrap_or_default() +} + +/// What the Media Player's output did with one frame. +pub struct Outcome { + /// The caps its `video_out` settled on, if it negotiated with the mixer. + pub negotiated: Option, + /// Every error on the bus, as (source name, message, debug). + pub errors: Vec<(String, String, String)>, +} + +/// Start only the Media Player's main-pipeline elements and push one frame +/// with `caps` from its `appsrc`, the way its bridge does with each decoded +/// sample. The mixer stays in NULL. Waits until the Media Player's output has +/// negotiated or the bus carries an error, at most 5 s. +pub fn push_from_player(pipeline: &gst::Pipeline, caps: &str) -> Outcome { + let names = ["appsrc_video", "queue_video", "video_out"].map(|n| format!("{}:{}", PLAYER, n)); + for name in names.iter().rev() { + let e = element(pipeline, name); + e.set_locked_state(true); + e.set_state(gst::State::Playing) + .unwrap_or_else(|e| panic!("{} could not start: {:?}", name, e)); + } + + let caps: gst::Caps = caps.parse().unwrap(); + let mut buffer = gst::Buffer::with_size(320 * 240 * 3 / 2).unwrap(); + buffer.get_mut().unwrap().set_pts(gst::ClockTime::ZERO); + let sample = gst::Sample::builder().buffer(&buffer).caps(&caps).build(); + let appsrc = element(pipeline, &names[0]) + .downcast::() + .unwrap(); + appsrc + .push_sample(&sample) + .expect("appsrc takes the sample"); + + let video_out_src = element(pipeline, &names[2]).static_pad("src").unwrap(); + let bus = pipeline.bus().unwrap(); + let mut errors = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(5); + let mut settle_until = None; + loop { + while let Some(msg) = bus.pop_filtered(&[gst::MessageType::Error]) { + if let gst::MessageView::Error(err) = msg.view() { + errors.push(( + msg.src().map(|s| s.name().to_string()).unwrap_or_default(), + err.error().to_string(), + err.debug().map(|d| d.to_string()).unwrap_or_default(), + )); + } + } + if video_out_src.current_caps().is_some() && settle_until.is_none() { + break; + } + // After the first error give the others (the Media Player's own + // not-negotiated) a moment to land too. + if !errors.is_empty() && settle_until.is_none() { + settle_until = Some(Instant::now() + Duration::from_millis(200)); + } + let now = Instant::now(); + if now >= deadline || settle_until.is_some_and(|t| now >= t) { + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + let negotiated = video_out_src.current_caps(); + + for name in &names { + let e = element(pipeline, name); + let _ = e.set_state(gst::State::Null); + e.set_locked_state(false); + } + Outcome { negotiated, errors } +} diff --git a/backend/tests/vision_mixer_cuda_input_test.rs b/backend/tests/vision_mixer_cuda_input_test.rs index b4ba2067e..249313719 100644 --- a/backend/tests/vision_mixer_cuda_input_test.rs +++ b/backend/tests/vision_mixer_cuda_input_test.rs @@ -8,42 +8,17 @@ //! element — in front of `glupload` when the producer offers CUDA memory only. //! //! These tests drive the real Media Player block into the real Vision Mixer -//! block. Only the Media Player's elements are started: the mixer stays in NULL, -//! so no GL context is needed — the queries that decide the splice happen while -//! the Media Player's `identity` settles its caps, before any CAPS event or -//! buffer reaches the mixer. The GL plugin must be installed (CI installs +//! block (see `vision_mixer_cuda`). The GL plugin must be installed (CI installs //! `gstreamer1.0-gl`); `cudadownload` is not, so a stand-in registered under -//! that name takes its place. Real CUDA memory needs an NVIDIA host. +//! that name takes its place. Real CUDA memory needs an NVIDIA host. The case +//! where `cudadownload` is missing is `vision_mixer_cuda_missing_test`. pub mod common; +pub mod vision_mixer_cuda; -use gstreamer as gst; use gstreamer::prelude::*; -use std::collections::HashMap; -use std::sync::Mutex; -use std::time::{Duration, Instant}; -use strom::blocks::BlockRegistry; -use strom::events::EventBroadcaster; use strom::gst::gl_input_front::CUDA_ADAPTER_FACTORY; -use strom::gst::pipeline::PipelineManager; -use strom_types::{Flow, PropertyValue as PV}; -use tempfile::NamedTempFile; - -/// The GL elements the GPU Vision Mixer is built from. Their absence is a CI -/// regression, not a reason to skip: a skip would pass green guarding nothing. -const GL_ELEMENTS: &[&str] = &["glupload", "glcolorconvert", "glvideomixerelement"]; - -const MIXER: &str = "vm"; -const PLAYER: &str = "player"; - -/// Both tests change what the registry offers under [`CUDA_ADAPTER_FACTORY`], -/// and cargo runs tests in one process in parallel. -static REGISTRY_LOCK: Mutex<()> = Mutex::new(()); - -const CUDA_CAPS: &str = "video/x-raw(memory:CUDAMemory),format=NV12,width=320,height=240,\ - framerate=25/1,pixel-aspect-ratio=1/1,interlace-mode=progressive"; -const SYSTEM_CAPS: &str = "video/x-raw,format=NV12,width=320,height=240,\ - framerate=25/1,pixel-aspect-ratio=1/1,interlace-mode=progressive"; +use vision_mixer_cuda::*; /// A stand-in for `cudadownload` with the same pad templates: CUDA or system /// memory in, GL or system memory out. The tests only need the element to @@ -128,190 +103,30 @@ mod stand_in { @extends gst::Element, gst::Object; } - /// Make [`super::CUDA_ADAPTER_FACTORY`] available for as long as the guard - /// lives. On an NVIDIA host the real element is left in place and used. - pub struct Registered(Option); - - pub fn register() -> Registered { - if gst::ElementFactory::find(super::CUDA_ADAPTER_FACTORY).is_some() { - return Registered(None); - } - gst::Element::register( - None, - super::CUDA_ADAPTER_FACTORY, - gst::Rank::NONE, - CudaDownloadStandIn::static_type(), - ) - .expect("register the cudadownload stand-in"); - Registered( - gst::ElementFactory::find(super::CUDA_ADAPTER_FACTORY) - .map(|f| f.upcast::()), - ) - } - - impl Drop for Registered { - fn drop(&mut self) { - if let Some(feature) = self.0.take() { - gst::Registry::get().remove_feature(&feature); - } - } - } -} - -fn block(id: &str, definition: &str, properties: Vec<(&str, PV)>) -> strom_types::BlockInstance { - let properties: HashMap = properties - .into_iter() - .map(|(k, v)| (k.to_string(), v)) - .collect(); - let computed_external_pads = strom::blocks::builtin::get_builder(definition) - .and_then(|builder| builder.get_external_pads(&properties)); - strom_types::BlockInstance { - id: id.to_string(), - block_definition_id: definition.to_string(), - name: None, - properties, - position: strom_types::block::Position { x: 0.0, y: 0.0 }, - runtime_data: None, - computed_external_pads, - } -} - -/// A Media Player (decode mode, empty playlist) feeding `mixer_input` of a GPU -/// Vision Mixer. -fn player_into_mixer(name: &str, mixer_input: &str) -> Flow { - let mut flow = Flow::new(name.to_string()); - flow.blocks.push(block( - MIXER, - "builtin.vision_mixer", - vec![ - ("compositor_preference", PV::String("gpu".to_string())), - ("num_inputs", PV::UInt(2)), - ("num_dsk_inputs", PV::String("1".to_string())), - ("pgm_resolution", PV::String("320x240".to_string())), - ("multiview_resolution", PV::String("320x240".to_string())), - ], - )); - flow.blocks.push(block( - PLAYER, - "builtin.media_player", - vec![("decode", PV::Bool(true))], - )); - flow.links.push(strom_types::Link { - from: format!("{}:video_out", PLAYER), - to: format!("{}:{}", MIXER, mixer_input), - }); - flow -} - -fn build_manager(flow: &Flow) -> (PipelineManager, NamedTempFile) { - gst::init().unwrap(); - strom::gpu::detect_gpu_capabilities(); - let temp_file = NamedTempFile::new().unwrap(); - let registry = BlockRegistry::new(temp_file.path()); - let manager = PipelineManager::new( - flow, - EventBroadcaster::with_capacity(10), - ®istry, - vec![], - "all".to_string(), - None, - std::env::temp_dir(), - std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())), - ) - .expect("GPU vision mixer flow should build"); - (manager, temp_file) -} - -fn element(pipeline: &gst::Pipeline, name: &str) -> gst::Element { - pipeline - .by_name(name) - .unwrap_or_else(|| panic!("{} exists", name)) -} - -/// The element feeding `name`'s sink pad. -fn feeder_of(pipeline: &gst::Pipeline, name: &str) -> Option { - element(pipeline, name) - .static_pad("sink")? - .peer()? - .parent_element() -} - -fn factory_name(element: &gst::Element) -> String { - element - .factory() - .map(|f| f.name().to_string()) - .unwrap_or_default() -} - -/// What the Media Player's output did with one frame. -struct Outcome { - /// The caps its `video_out` settled on, if it negotiated with the mixer. - negotiated: Option, - /// Every error on the bus, as (source name, message, debug). - errors: Vec<(String, String, String)>, -} - -/// Start only the Media Player's main-pipeline elements and push one frame -/// with `caps` from its `appsrc`, the way its bridge does with each decoded -/// sample. The mixer stays in NULL. Waits until the Media Player's output has -/// negotiated or the bus carries an error, at most 5 s. -fn push_from_player(pipeline: &gst::Pipeline, caps: &str) -> Outcome { - let names = ["appsrc_video", "queue_video", "video_out"].map(|n| format!("{}:{}", PLAYER, n)); - for name in names.iter().rev() { - let e = element(pipeline, name); - e.set_locked_state(true); - e.set_state(gst::State::Playing) - .unwrap_or_else(|e| panic!("{} could not start: {:?}", name, e)); - } - - let caps: gst::Caps = caps.parse().unwrap(); - let mut buffer = gst::Buffer::with_size(320 * 240 * 3 / 2).unwrap(); - buffer.get_mut().unwrap().set_pts(gst::ClockTime::ZERO); - let sample = gst::Sample::builder().buffer(&buffer).caps(&caps).build(); - let appsrc = element(pipeline, &names[0]) - .downcast::() - .unwrap(); - appsrc - .push_sample(&sample) - .expect("appsrc takes the sample"); - - let video_out_src = element(pipeline, &names[2]).static_pad("src").unwrap(); - let bus = pipeline.bus().unwrap(); - let mut errors = Vec::new(); - let deadline = Instant::now() + Duration::from_secs(5); - let mut settle_until = None; - loop { - while let Some(msg) = bus.pop_filtered(&[gst::MessageType::Error]) { - if let gst::MessageView::Error(err) = msg.view() { - errors.push(( - msg.src().map(|s| s.name().to_string()).unwrap_or_default(), - err.error().to_string(), - err.debug().map(|d| d.to_string()).unwrap_or_default(), - )); + /// Make [`super::CUDA_ADAPTER_FACTORY`] available for the rest of the + /// process. On an NVIDIA host the real element is left in place and used. + /// + /// Registered once and never removed: `Registry::remove_feature` frees the + /// factory while the element class keeps a plain pointer to it, so a later + /// lookup or registration under the same name reads freed memory. That + /// crashed this binary in about one run in four. The missing-adapter case + /// lives in its own test binary instead, where nothing is registered. + pub fn register() { + static ONCE: std::sync::Once = std::sync::Once::new(); + ONCE.call_once(|| { + gst::init().expect("GStreamer initialises"); + if gst::ElementFactory::find(super::CUDA_ADAPTER_FACTORY).is_some() { + return; } - } - if video_out_src.current_caps().is_some() && settle_until.is_none() { - break; - } - // After the first error give the others (the Media Player's own - // not-negotiated) a moment to land too. - if !errors.is_empty() && settle_until.is_none() { - settle_until = Some(Instant::now() + Duration::from_millis(200)); - } - let now = Instant::now(); - if now >= deadline || settle_until.is_some_and(|t| now >= t) { - break; - } - std::thread::sleep(Duration::from_millis(20)); - } - let negotiated = video_out_src.current_caps(); - - for name in &names { - let e = element(pipeline, name); - let _ = e.set_state(gst::State::Null); - e.set_locked_state(false); + gst::Element::register( + None, + super::CUDA_ADAPTER_FACTORY, + gst::Rank::NONE, + CudaDownloadStandIn::static_type(), + ) + .expect("register the cudadownload stand-in"); + }); } - Outcome { negotiated, errors } } /// The defect: CUDA memory from the Media Player into a mixer input. With the @@ -320,9 +135,8 @@ fn push_from_player(pipeline: &gst::Pipeline, caps: &str) -> Outcome { /// Media Player failed with not-negotiated. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cuda_memory_from_the_media_player_gets_cudadownload_before_glupload() { - let _lock = REGISTRY_LOCK.lock().unwrap_or_else(|p| p.into_inner()); common::require_elements(GL_ELEMENTS); - let _adapter = stand_in::register(); + stand_in::register(); for input in ["video_in_0", "dsk_in_0"] { let suffix = if input == "dsk_in_0" { "dsk_0" } else { "0" }; @@ -386,9 +200,8 @@ async fn cuda_memory_from_the_media_player_gets_cudadownload_before_glupload() { /// into `glupload`, which uploads it. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn system_memory_from_the_media_player_goes_straight_into_glupload() { - let _lock = REGISTRY_LOCK.lock().unwrap_or_else(|p| p.into_inner()); common::require_elements(GL_ELEMENTS); - let _adapter = stand_in::register(); + stand_in::register(); let (manager, _registry) = build_manager(&player_into_mixer("system_into_vm", "video_in_0")); let pipeline = manager.pipeline(); @@ -405,45 +218,3 @@ async fn system_memory_from_the_media_player_goes_straight_into_glupload() { "a system-memory input was given an adapter it does not need" ); } - -/// Without `cudadownload` (no nvcodec plugin), CUDA memory must fail the flow -/// with a message that names what is missing, posted by the mixer input — -/// not only the Media Player's `Internal data stream error`. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn cuda_memory_without_cudadownload_fails_the_flow_with_a_reason() { - let _lock = REGISTRY_LOCK.lock().unwrap_or_else(|p| p.into_inner()); - common::require_elements(GL_ELEMENTS); - if gst::ElementFactory::find(CUDA_ADAPTER_FACTORY).is_some() { - // Only an NVIDIA build has the real element; CI never does. - eprintln!( - "SKIP: {} is installed, so the missing-adapter path cannot be reached here", - CUDA_ADAPTER_FACTORY - ); - return; - } - - let (manager, _registry) = build_manager(&player_into_mixer("cuda_no_adapter", "video_in_0")); - let pipeline = manager.pipeline(); - let queue = format!("{}:queue_0", MIXER); - let errors = push_from_player(pipeline, CUDA_CAPS).errors; - let ours = errors.iter().find(|(source, _, _)| *source == queue); - let Some((_, message, debug)) = ours else { - panic!( - "no error from {}: CUDA memory without {} fails the flow without saying \ - why. Errors on the bus: {:?}", - queue, CUDA_ADAPTER_FACTORY, errors - ); - }; - assert!( - message.contains(CUDA_ADAPTER_FACTORY) - && message.contains("CUDA memory") - && message.contains("video_in_0"), - "the error does not say which input lacks what: {}", - message - ); - assert!( - debug.contains("nvcodec"), - "the error does not say how to fix it: {}", - debug - ); -} diff --git a/backend/tests/vision_mixer_cuda_missing_test.rs b/backend/tests/vision_mixer_cuda_missing_test.rs new file mode 100644 index 000000000..7918e620c --- /dev/null +++ b/backend/tests/vision_mixer_cuda_missing_test.rs @@ -0,0 +1,55 @@ +//! Regression test: CUDA memory into the GPU Vision Mixer on a host without +//! `cudadownload` fails the flow with a reason (#682). +//! +//! Its own test binary, so no stand-in `cudadownload` is ever registered in +//! this process: `vision_mixer_cuda_input_test` registers one for the life of +//! its process, and a GStreamer registry feature cannot be safely removed. + +pub mod common; +pub mod vision_mixer_cuda; + +use gstreamer as gst; +use strom::gst::gl_input_front::CUDA_ADAPTER_FACTORY; +use vision_mixer_cuda::*; + +/// Without `cudadownload` (no nvcodec plugin), CUDA memory must fail the flow +/// with a message that names what is missing, posted by the mixer input — +/// not only the Media Player's `Internal data stream error`. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cuda_memory_without_cudadownload_fails_the_flow_with_a_reason() { + common::require_elements(GL_ELEMENTS); + gst::init().unwrap(); + if gst::ElementFactory::find(CUDA_ADAPTER_FACTORY).is_some() { + // Only an NVIDIA build has the real element; CI never does. + eprintln!( + "SKIP: {} is installed, so the missing-adapter path cannot be reached here", + CUDA_ADAPTER_FACTORY + ); + return; + } + + let (manager, _registry) = build_manager(&player_into_mixer("cuda_no_adapter", "video_in_0")); + let pipeline = manager.pipeline(); + let queue = format!("{}:queue_0", MIXER); + let errors = push_from_player(pipeline, CUDA_CAPS).errors; + let ours = errors.iter().find(|(source, _, _)| *source == queue); + let Some((_, message, debug)) = ours else { + panic!( + "no error from {}: CUDA memory without {} fails the flow without saying \ + why. Errors on the bus: {:?}", + queue, CUDA_ADAPTER_FACTORY, errors + ); + }; + assert!( + message.contains(CUDA_ADAPTER_FACTORY) + && message.contains("CUDA memory") + && message.contains("video_in_0"), + "the error does not say which input lacks what: {}", + message + ); + assert!( + debug.contains("nvcodec"), + "the error does not say how to fix it: {}", + debug + ); +}