diff --git a/.github/workflows/sync-tests.yml b/.github/workflows/sync-tests.yml index 2fe441dd908..e16dd232809 100644 --- a/.github/workflows/sync-tests.yml +++ b/.github/workflows/sync-tests.yml @@ -23,6 +23,8 @@ on: - "crates/enc-mediafoundation/**" - "crates/timestamp/**" - "crates/rendering/**" + - "crates/video-decode/**" + - "crates/d3d-adapter/**" - "crates/render-farm/**" - "crates/editor/**" - "crates/export/**" @@ -156,6 +158,19 @@ jobs: cargo test --locked -p cap-rendering -- --nocapture --skip zoom_spring::tests::precompute_cost_is_bounded_for_long_projects --skip decoder::model:: cargo test --locked -p cap-rendering --lib zoom_spring::tests::precompute_cost_is_bounded_for_long_projects -- --exact --nocapture --test-threads=1 + # The hosted runner has no video-capable D3D11 adapter, so these tests + # force plain WARP and exercise the software Media Foundation path. The + # required flags turn a missing device or decoder into a failure instead + # of a silent skip. + - name: Media Foundation decoder tests (Windows) + if: runner.os == 'Windows' + shell: bash + env: + CAP_FORCE_WARP_TESTS: "1" + CAP_REQUIRE_D3D11_TESTS: "1" + CAP_REQUIRE_MF_TESTS: "1" + run: cargo test --locked -p cap-video-decode --lib -- --nocapture --test-threads=1 + - name: Editor audio playback and export regressions shell: bash run: | diff --git a/crates/rendering/src/decoder/media_foundation.rs b/crates/rendering/src/decoder/media_foundation.rs index 9081af17c58..332df3e6e3b 100644 --- a/crates/rendering/src/decoder/media_foundation.rs +++ b/crates/rendering/src/decoder/media_foundation.rs @@ -192,13 +192,14 @@ impl MFDecoder { ) }, |data| { - DecodedFrame::new_nv12_with_arc( + DecodedFrame::new_nv12_with_mapped( data.data, video_width, video_height, data.y_stride, data.uv_stride, ) + .with_storage_epoch(data.storage_epoch) .with_matrix(matrix) }, )); @@ -246,9 +247,6 @@ impl MFDecoder { Step::Seek(frame) => match decoder.seek(frame_to_100ns(frame, fps)) { Ok(()) => { requests.seeked(frame); - for retired in requests.take_retired() { - decoder.recycle_nv12_buffer(retired.data); - } Vec::new() } Err(e) => { @@ -262,15 +260,11 @@ impl MFDecoder { Ok(Some(sample)) => { health.record_success(decode_start.elapsed()); let bytes = sample.data.data.len(); - let answers = requests.decoded( + requests.decoded( pts_100ns_to_frame(sample.pts, fps), sample.data, bytes, - ); - for retired in requests.take_retired() { - decoder.recycle_nv12_buffer(retired.data); - } - answers + ) } Ok(None) => requests.stopped(), Err(e) => { @@ -324,7 +318,6 @@ struct FrameRequests { // below a frame is the stream's newest sample there. cache: BTreeMap, cache_bytes: usize, - retired: Vec, waiting: Vec<(u32, R)>, seeked_to: Option, readahead_frames: u32, @@ -336,7 +329,6 @@ impl FrameRequests { Self { cache: BTreeMap::new(), cache_bytes: 0, - retired: Vec::new(), waiting: Vec::new(), seeked_to: None, readahead_frames, @@ -411,19 +403,11 @@ impl FrameRequests { } fn seeked(&mut self, frame: u32) { - self.retired.extend( - std::mem::take(&mut self.cache) - .into_values() - .map(|(sample, _)| sample), - ); + self.cache.clear(); self.cache_bytes = 0; self.seeked_to = Some(frame); } - fn take_retired(&mut self) -> Vec { - std::mem::take(&mut self.retired) - } - // A waiting frame the decoder jumped past has no sample at or below it since the seek, so // it gets the first one after. fn decoded(&mut self, frame: u32, sample: T, bytes: usize) -> Vec> { @@ -439,17 +423,15 @@ impl FrameRequests { }) .collect(); - if let Some((replaced, bytes)) = self.cache.remove(&frame) { - self.retired.push(replaced); - self.cache_bytes -= bytes; + if let Some((_, replaced)) = self.cache.remove(&frame) { + self.cache_bytes -= replaced; } while !self.cache.is_empty() && (self.cache.len() >= FRAME_CACHE_SIZE || self.cache_bytes + bytes > MAX_FRAME_CACHE_BYTES) { - if let Some((_, (evicted, bytes))) = self.cache.pop_first() { - self.retired.push(evicted); - self.cache_bytes -= bytes; + if let Some((_, (_, evicted))) = self.cache.pop_first() { + self.cache_bytes -= evicted; } } self.cache.insert(frame, (sample, bytes)); @@ -488,20 +470,6 @@ mod tests { assert_eq!(pts_100ns_to_frame(10_000_000, 30), 30); } - #[test] - fn cache_replacements_evictions_and_seeks_retire_samples() { - let mut requests = FrameRequests::::new(0); - requests.decoded(0, 1, 1); - requests.decoded(0, 2, 1); - assert_eq!(requests.take_retired(), [1]); - - requests.decoded(1, 3, MAX_FRAME_CACHE_BYTES); - assert_eq!(requests.take_retired(), [2]); - - requests.seeked(10); - assert_eq!(requests.take_retired(), [3]); - } - // Samples are their index in `frames`, which holds the frame each one rounds to. struct FakeStream { frames: Vec, diff --git a/crates/rendering/src/decoder/mod.rs b/crates/rendering/src/decoder/mod.rs index b8c827239b1..e5f475c8fcc 100644 --- a/crates/rendering/src/decoder/mod.rs +++ b/crates/rendering/src/decoder/mod.rs @@ -152,10 +152,13 @@ pub struct DecodedFrame { // its planes into `data`. Layers use the backing as the per-frame storage // identity so re-requesting the same decoded frame skips the upload. ffmpeg_backing: Option>, + storage_epoch: u64, #[cfg(target_os = "macos")] image_buf_backing: Option>, #[cfg(target_os = "windows")] d3d11_texture_backing: Option>, + #[cfg(target_os = "windows")] + mapped_nv12: Option>, #[cfg(target_os = "linux")] cuda_nv12: Option>, } @@ -243,11 +246,15 @@ pub(crate) struct DecodedFrameStorageIdentity { format: PixelFormat, y_stride: u32, uv_stride: u32, + storage_epoch: u64, } impl DecodedFrameStorageIdentity { pub(crate) fn matches(&self, other: &Self) -> bool { - let same_storage = match (&self.data, &other.data) { + // Mapped staging textures are recycled, so a frame built from the same + // `data` slot can hold different pixels; the readback epoch identifies + // the actual sample and must win when either side has one. + let same_data = match (&self.data, &other.data) { (Some(data), Some(other_data)) => data.ptr_eq(other_data), (None, None) => true, _ => false, @@ -257,6 +264,11 @@ impl DecodedFrameStorageIdentity { (None, None) => true, _ => false, }; + let same_storage = if self.storage_epoch != 0 || other.storage_epoch != 0 { + self.storage_epoch == other.storage_epoch + } else { + same_data + }; same_storage && same_ffmpeg_backing && self.width == other.width @@ -290,11 +302,14 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride: width * 4, uv_stride: 0, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -309,11 +324,14 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride: width * 4, uv_stride: 0, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -335,6 +353,7 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride: frame.y_pitch as u32, uv_stride: frame.uv_pitch as u32, + storage_epoch: 0, ffmpeg_backing: None, cuda_nv12: Some(frame), } @@ -354,11 +373,14 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride, uv_stride, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -379,11 +401,14 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride, uv_stride, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -406,10 +431,13 @@ impl DecodedFrame { y_stride, uv_stride, ffmpeg_backing: Some(frame), + storage_epoch: 0, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -442,6 +470,37 @@ impl DecodedFrame { ) } + #[cfg(target_os = "windows")] + pub(crate) fn new_nv12_with_mapped( + mapped: Arc, + width: u32, + height: u32, + y_stride: u32, + uv_stride: u32, + ) -> Self { + Self { + data: Arc::new(Vec::new()), + width, + height, + format: PixelFormat::Nv12, + matrix: YuvMatrix::Bt601, + y_stride, + uv_stride, + ffmpeg_backing: None, + storage_epoch: 0, + #[cfg(target_os = "macos")] + image_buf_backing: None, + d3d11_texture_backing: None, + mapped_nv12: Some(mapped), + } + } + + #[cfg(target_os = "windows")] + pub(crate) fn with_storage_epoch(mut self, storage_epoch: u64) -> Self { + self.storage_epoch = storage_epoch; + self + } + #[cfg(target_os = "macos")] pub(crate) fn new_nv12_with_image_buf( width: u32, @@ -458,6 +517,7 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride, uv_stride, + storage_epoch: 0, ffmpeg_backing: None, image_buf_backing: Some(Arc::new(SendableImageBuf::new(image_buf))), } @@ -478,11 +538,14 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride, uv_stride, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -503,11 +566,14 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride, uv_stride, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, #[cfg(target_os = "windows")] d3d11_texture_backing: None, + #[cfg(target_os = "windows")] + mapped_nv12: None, #[cfg(target_os = "linux")] cuda_nv12: None, } @@ -523,10 +589,12 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride: width, uv_stride: width, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, d3d11_texture_backing: Some(Arc::new(SendableD3D11Texture::new(texture))), + mapped_nv12: None, } } @@ -545,6 +613,7 @@ impl DecodedFrame { matrix: YuvMatrix::Bt601, y_stride: width, uv_stride: width, + storage_epoch: 0, ffmpeg_backing: None, #[cfg(target_os = "macos")] image_buf_backing: None, @@ -552,6 +621,7 @@ impl DecodedFrame { texture, shared_handle, ))), + mapped_nv12: None, } } @@ -586,8 +656,14 @@ impl DecodedFrame { /// /// Borrowed FFmpeg YUV frames expose their planes through [`Self::y_plane`], /// [`Self::uv_plane`], [`Self::u_plane`] and [`Self::v_plane`] instead and - /// return an empty slice here. + /// return an empty slice here. Mapped Media Foundation frames return their + /// mapped NV12 buffer. pub fn data(&self) -> &[u8] { + #[cfg(target_os = "windows")] + if let Some(mapped) = &self.mapped_nv12 { + return mapped.as_slice(); + } + #[cfg(target_os = "macos")] if self.data.is_empty() && let Some(backing) = &self.image_buf_backing @@ -628,6 +704,11 @@ impl DecodedFrame { } pub fn byte_len(&self) -> usize { + #[cfg(target_os = "windows")] + if let Some(mapped) = &self.mapped_nv12 { + return mapped.len(); + } + if !self.data.is_empty() { return self.data.len(); } @@ -658,6 +739,7 @@ impl DecodedFrame { format: self.format, y_stride: self.y_stride, uv_stride: self.uv_stride, + storage_epoch: self.storage_epoch, } } @@ -2056,6 +2138,25 @@ mod tests { ); } + #[test] + fn staging_epoch_distinguishes_recycled_slots() { + let data = Arc::new(vec![0; 8]); + let identity = |storage_epoch| DecodedFrameStorageIdentity { + data: Some(Arc::downgrade(&data)), + ffmpeg_backing: None, + width: 2, + height: 2, + format: PixelFormat::Nv12, + y_stride: 2, + uv_stride: 2, + storage_epoch, + }; + + assert!(identity(7).matches(&identity(7))); + assert!(!identity(7).matches(&identity(8))); + assert!(!identity(7).matches(&identity(0))); + } + #[test] fn ffmpeg_backed_storage_identity_follows_the_shared_planes() { let width = 4; @@ -2111,6 +2212,48 @@ mod tests { ); } + #[test] + fn storage_identity_combines_ffmpeg_backing_and_staging_epochs() { + let data = Arc::new(vec![0; 8]); + let backing = Arc::new(::ffmpeg::frame::Video::new( + ::ffmpeg::format::Pixel::NV12, + 2, + 2, + )); + let mapped = |storage_epoch| DecodedFrameStorageIdentity { + data: Some(Arc::downgrade(&data)), + ffmpeg_backing: None, + width: 2, + height: 2, + format: PixelFormat::Nv12, + y_stride: 2, + uv_stride: 2, + storage_epoch, + }; + let borrowed = |backing: &Arc<::ffmpeg::frame::Video>| DecodedFrameStorageIdentity { + data: None, + ffmpeg_backing: Some(Arc::downgrade(backing)), + width: 2, + height: 2, + format: PixelFormat::Nv12, + y_stride: 2, + uv_stride: 2, + storage_epoch: 0, + }; + + assert!(mapped(3).matches(&mapped(3))); + assert!(!mapped(3).matches(&mapped(4))); + assert!(!mapped(3).matches(&borrowed(&backing))); + assert!(borrowed(&backing).matches(&borrowed(&backing))); + assert!( + !borrowed(&backing).matches(&borrowed(&Arc::new(::ffmpeg::frame::Video::new( + ::ffmpeg::format::Pixel::NV12, + 2, + 2 + )))) + ); + } + #[cfg(target_os = "macos")] #[test] fn iosurface_backed_nv12_materializes_raw_planes_only_on_demand() { diff --git a/crates/video-decode/src/lib.rs b/crates/video-decode/src/lib.rs index 9fb59896186..2d935dd87ca 100644 --- a/crates/video-decode/src/lib.rs +++ b/crates/video-decode/src/lib.rs @@ -11,6 +11,6 @@ pub use avassetreader::{AVAssetReaderDecoder, KeyframeIndex}; pub use ffmpeg::FFmpegDecoder; #[cfg(target_os = "windows")] pub use media_foundation::{ - FrameTextures, MFDecodedFrame, MFDecoderCapabilities, MediaFoundationDecoder, NV12Data, - NV12Sample, get_mf_decoder_capabilities, + FrameTextures, MFDecodedFrame, MFDecoderCapabilities, MappedNV12, MediaFoundationDecoder, + NV12Data, NV12ReadbackStats, NV12Sample, get_mf_decoder_capabilities, }; diff --git a/crates/video-decode/src/media_foundation.rs b/crates/video-decode/src/media_foundation.rs index a4256fb01d7..56cb301b175 100644 --- a/crates/video-decode/src/media_foundation.rs +++ b/crates/video-decode/src/media_foundation.rs @@ -1,6 +1,7 @@ use std::collections::VecDeque; use std::path::Path; -use std::sync::{Arc, OnceLock}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, OnceLock, mpsc}; use std::time::{Duration, Instant}; use tracing::{info, warn}; #[cfg(target_os = "windows")] @@ -10,26 +11,29 @@ use windows::{ Win32::{ Foundation::{HANDLE, HMODULE}, Graphics::{ - Direct3D::{ - D3D_DRIVER_TYPE_WARP, D3D_FEATURE_LEVEL, D3D_SRV_DIMENSION_TEXTURE2D, ID3DBlob, - }, + Direct3D::{D3D_FEATURE_LEVEL, D3D_SRV_DIMENSION_TEXTURE2D, ID3DBlob}, Direct3D11::{ D3D11_BIND_CONSTANT_BUFFER, D3D11_BIND_SHADER_RESOURCE, - D3D11_BIND_UNORDERED_ACCESS, D3D11_BUFFER_DESC, D3D11_CPU_ACCESS_READ, + D3D11_BIND_UNORDERED_ACCESS, D3D11_BOX, D3D11_BUFFER_DESC, D3D11_CPU_ACCESS_READ, D3D11_CREATE_DEVICE_BGRA_SUPPORT, D3D11_CREATE_DEVICE_VIDEO_SUPPORT, D3D11_DECODER_PROFILE_H264_VLD_NOFGT, D3D11_DECODER_PROFILE_HEVC_VLD_MAIN, - D3D11_MAP_READ, D3D11_MAPPED_SUBRESOURCE, D3D11_SDK_VERSION, - D3D11_SHADER_RESOURCE_VIEW_DESC1, D3D11_SHADER_RESOURCE_VIEW_DESC1_0, - D3D11_TEX2D_SRV1, D3D11_TEX2D_UAV, D3D11_TEXTURE2D_DESC, - D3D11_UAV_DIMENSION_TEXTURE2D, D3D11_UNORDERED_ACCESS_VIEW_DESC, - D3D11_UNORDERED_ACCESS_VIEW_DESC_0, D3D11_USAGE_DEFAULT, D3D11_USAGE_STAGING, - D3D11_VIDEO_DECODER_DESC, D3D11CreateDevice, ID3D11Buffer, ID3D11ComputeShader, - ID3D11Device, ID3D11Device3, ID3D11DeviceContext, ID3D11DeviceContext1, - ID3D11ShaderResourceView, ID3D11ShaderResourceView1, ID3D11Texture2D, - ID3D11UnorderedAccessView, ID3D11VideoDevice, + D3D11_MAP_FLAG_DO_NOT_WAIT, D3D11_MAP_READ, D3D11_MAPPED_SUBRESOURCE, + D3D11_SDK_VERSION, D3D11_SHADER_RESOURCE_VIEW_DESC1, + D3D11_SHADER_RESOURCE_VIEW_DESC1_0, D3D11_TEX2D_SRV1, D3D11_TEX2D_UAV, + D3D11_TEXTURE2D_DESC, D3D11_UAV_DIMENSION_TEXTURE2D, + D3D11_UNORDERED_ACCESS_VIEW_DESC, D3D11_UNORDERED_ACCESS_VIEW_DESC_0, + D3D11_USAGE_DEFAULT, D3D11_USAGE_STAGING, D3D11_VIDEO_DECODER_DESC, + D3D11CreateDevice, ID3D11Buffer, ID3D11ComputeShader, ID3D11Device, ID3D11Device3, + ID3D11DeviceContext, ID3D11DeviceContext1, ID3D11ShaderResourceView, + ID3D11ShaderResourceView1, ID3D11Texture2D, ID3D11UnorderedAccessView, + ID3D11VideoDevice, }, - Dxgi::Common::{ - DXGI_FORMAT_NV12, DXGI_FORMAT_R8_UNORM, DXGI_FORMAT_R8G8_UNORM, DXGI_SAMPLE_DESC, + Dxgi::{ + Common::{ + DXGI_FORMAT_NV12, DXGI_FORMAT_R8_UNORM, DXGI_FORMAT_R8G8_UNORM, + DXGI_SAMPLE_DESC, + }, + DXGI_ERROR_WAS_STILL_DRAWING, }, }, Media::MediaFoundation::{ @@ -161,9 +165,18 @@ pub struct NV12Sample { #[derive(Clone)] pub struct NV12Data { - pub data: Arc>, + pub data: Arc, pub y_stride: u32, pub uv_stride: u32, + /// Identifies this fill of a reused staging texture, so storage-identity + /// comparisons do not mistake a recycled allocation for the same frame. + pub storage_epoch: u64, +} + +impl NV12Data { + pub fn as_slice(&self) -> &[u8] { + self.data.as_slice() + } } #[derive(Clone)] pub struct FrameTexture { @@ -583,61 +596,319 @@ fn create_uav( } } -// Samples whose copy to a staging texture can be in flight while an earlier one is read back. -const READBACK_DEPTH: usize = 2; -const MAX_CPU_READBACK_BUFFERS: usize = 8; -const MAX_CPU_READBACK_BUFFER_BYTES: usize = 128 * 1024 * 1024; +const MAX_IDLE_STAGING_BYTES: usize = 32 * 1024 * 1024; +// Idle readback textures are process-wide: per-decoder caps would multiply the +// staging footprint in projects with many segment and transition decoders. +const MAX_GLOBAL_IDLE_STAGING_BYTES: usize = 128 * 1024 * 1024; +static NEXT_READBACK_EPOCH: AtomicU64 = AtomicU64::new(1); +static GLOBAL_IDLE_STAGING_BYTES: AtomicUsize = AtomicUsize::new(0); + +fn reserve_global_staging(bytes: usize) -> bool { + let mut current = GLOBAL_IDLE_STAGING_BYTES.load(Ordering::Relaxed); + loop { + let Some(next) = current.checked_add(bytes) else { + return false; + }; + if next > MAX_GLOBAL_IDLE_STAGING_BYTES { + return false; + } + match GLOBAL_IDLE_STAGING_BYTES.compare_exchange_weak( + current, + next, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => return true, + Err(actual) => current = actual, + } + } +} #[derive(Default)] -struct CpuReadbackPool { - free: Vec>, - pending: Vec>>, +struct FreeStagingTextures { + textures: Vec, + bytes: usize, +} + +struct StagingTexturePool { + free: std::sync::Mutex, + device: ID3D11Device, + width: u32, + height: u32, + frame_bytes: usize, } -impl CpuReadbackPool { - fn acquire(&mut self, size: usize) -> Vec { - self.collect(); - let mut buffer = self.free.pop().unwrap_or_default(); - buffer.clear(); - if buffer.capacity() < size { - buffer.reserve(size); +unsafe impl Send for StagingTexturePool {} +unsafe impl Sync for StagingTexturePool {} + +impl StagingTexturePool { + fn new(device: ID3D11Device, width: u32, height: u32) -> Self { + Self { + free: std::sync::Mutex::new(FreeStagingTextures::default()), + device, + width, + height, + frame_bytes: (width as usize) + .saturating_mul(height as usize) + .saturating_mul(3) + / 2, } - buffer } - fn recycle(&mut self, buffer: Arc>) { - self.collect(); - match Arc::try_unwrap(buffer) { - Ok(buffer) if self.can_retain(buffer.capacity()) => self.free.push(buffer), - Ok(_) => {} - Err(buffer) if self.can_retain(buffer.capacity()) => self.pending.push(buffer), - Err(_) => {} + fn acquire(&self) -> Result<(ID3D11Texture2D, bool), String> { + let mut free = self + .free + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(texture) = free.textures.pop() { + free.bytes = free.bytes.saturating_sub(self.frame_bytes); + let _ = GLOBAL_IDLE_STAGING_BYTES.fetch_update( + Ordering::Relaxed, + Ordering::Relaxed, + |bytes| Some(bytes.saturating_sub(self.frame_bytes)), + ); + return Ok((texture, true)); + } + drop(free); + + let texture = unsafe { create_staging_texture(&self.device, self.width, self.height)? }; + Ok((texture, false)) + } + + fn recycle(&self, texture: ID3D11Texture2D) { + let mut free = self + .free + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if free.bytes.saturating_add(self.frame_bytes) > MAX_IDLE_STAGING_BYTES + || !reserve_global_staging(self.frame_bytes) + { + return; + } + + free.textures.push(texture); + free.bytes = free.bytes.saturating_add(self.frame_bytes); + } +} + +impl Drop for StagingTexturePool { + fn drop(&mut self) { + let free = self + .free + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if free.bytes > 0 { + let _ = GLOBAL_IDLE_STAGING_BYTES.fetch_update( + Ordering::Relaxed, + Ordering::Relaxed, + |bytes| Some(bytes.saturating_sub(free.bytes)), + ); + } + } +} + +struct ReclaimTexture { + texture: Option, + context: ID3D11DeviceContext, +} + +// SAFETY: the decoder's immediate context has multithread protection enabled, +// and the texture is only touched again after the decoder drains it. +unsafe impl Send for ReclaimTexture {} + +impl ReclaimTexture { + fn new(texture: ID3D11Texture2D, context: ID3D11DeviceContext) -> Self { + Self { + texture: Some(texture), + context, } } - fn collect(&mut self) { - for buffer in std::mem::take(&mut self.pending) { - match Arc::try_unwrap(buffer) { - Ok(buffer) => self.free.push(buffer), - Err(buffer) => self.pending.push(buffer), + fn take(&mut self) -> Option { + self.texture.take() + } +} + +impl Drop for ReclaimTexture { + fn drop(&mut self) { + // The decoder can drop its receiver with reclaims still queued, so + // unmap here on the decoder's thread before releasing the context. + if let Some(texture) = self.texture.take() { + unsafe { + self.context.Unmap(&texture, 0); } } } +} + +/// A decoded NV12 frame left mapped in its staging texture. The mapping stays +/// valid until the last clone is dropped, so plane access involves no copy. +pub struct MappedNV12 { + texture: Option, + reclaim: mpsc::Sender, + context: ID3D11DeviceContext, + _device: ID3D11Device, + pool: Arc, + _init_guard: Arc, + ptr: *mut u8, + len: usize, +} + +// SAFETY: the mapped bytes are immutable while shared, and the last reference +// hands the texture back to the decoder's thread before it is reused. +unsafe impl Send for MappedNV12 {} +unsafe impl Sync for MappedNV12 {} + +impl MappedNV12 { + pub fn as_slice(&self) -> &[u8] { + unsafe { std::slice::from_raw_parts(self.ptr, self.len) } + } - fn can_retain(&self, capacity: usize) -> bool { - self.free.len() + self.pending.len() < MAX_CPU_READBACK_BUFFERS - && self.retained_bytes().saturating_add(capacity) <= MAX_CPU_READBACK_BUFFER_BYTES + pub fn len(&self) -> usize { + self.len } - fn retained_bytes(&self) -> usize { - self.free - .iter() - .map(Vec::capacity) - .chain(self.pending.iter().map(|buffer| buffer.capacity())) - .fold(0, |total, capacity| total.saturating_add(capacity)) + pub fn is_empty(&self) -> bool { + self.len == 0 } } +impl Drop for MappedNV12 { + fn drop(&mut self) { + let Some(texture) = self.texture.take() else { + return; + }; + + if let Err(mut error) = self + .reclaim + .send(ReclaimTexture::new(texture, self.context.clone())) + { + // The decoder has shut down, so there is no thread that can still + // hold the device lock; unmap before releasing the texture. + if let Some(texture) = error.0.take() { + unsafe { + self.context.Unmap(&texture, 0); + } + self.pool.recycle(texture); + } + } + } +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct NV12ReadbackStats { + pub frames: u64, + pub bytes: u64, + pub map_immediate: u64, + pub map_waited: u64, + pub map_blocked: u64, + pub map_wait: Duration, + pub pool_hits: u64, + pub pool_misses: u64, +} + +impl NV12ReadbackStats { + fn record_frame(&mut self, bytes: usize, map_wait: Duration, attempts: u32, blocked: bool) { + self.frames += 1; + self.bytes = self.bytes.saturating_add(bytes as u64); + self.map_wait += map_wait; + if attempts == 0 { + self.map_immediate += 1; + } else { + self.map_waited += 1; + } + if blocked { + self.map_blocked += 1; + } + + if self.frames % 240 == 0 { + tracing::debug!( + frames = self.frames, + readback_mb = self.bytes as f64 / (1024.0 * 1024.0), + map_immediate = self.map_immediate, + map_waited = self.map_waited, + map_blocked = self.map_blocked, + map_wait_us = self.map_wait.as_secs_f64() * 1_000_000.0 / self.frames as f64, + pool_hits = self.pool_hits, + pool_misses = self.pool_misses, + "MediaFoundation NV12 readback" + ); + } + } +} + +unsafe fn map_nv12_staging( + context: &ID3D11DeviceContext, + staging: ID3D11Texture2D, + height: u32, + pool: &Arc, + reclaim: &mpsc::Sender, + stats: &mut NV12ReadbackStats, + init_guard: Arc, +) -> Result { + let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + let map_start = Instant::now(); + let mut attempts = 0u32; + let mut blocked = false; + + match unsafe { + context.Map( + &staging, + 0, + D3D11_MAP_READ, + D3D11_MAP_FLAG_DO_NOT_WAIT.0 as u32, + Some(&mut mapped), + ) + } { + Ok(()) => {} + Err(error) if error.code() == DXGI_ERROR_WAS_STILL_DRAWING => { + attempts = 1; + // The copy is still in flight and this thread has no other work to + // overlap, so spinning would only add to the wait. Block in Map. + if let Err(error) = + unsafe { context.Map(&staging, 0, D3D11_MAP_READ, 0, Some(&mut mapped)) } + { + pool.recycle(staging); + return Err(format!("Map staging texture failed: {error:?}")); + } + blocked = true; + } + Err(error) => { + pool.recycle(staging); + return Err(format!("Map staging texture failed: {error:?}")); + } + } + let map_wait = map_start.elapsed(); + + let y_stride = mapped.RowPitch; + let uv_height = height.div_ceil(2); + let len = (y_stride as usize) + .saturating_mul(height as usize) + .saturating_add((y_stride as usize).saturating_mul(uv_height as usize)); + let storage_epoch = NEXT_READBACK_EPOCH.fetch_add(1, Ordering::Relaxed); + + stats.record_frame(len, map_wait, attempts, blocked); + + Ok(NV12Data { + data: Arc::new(MappedNV12 { + texture: Some(staging), + reclaim: reclaim.clone(), + context: context.clone(), + _device: pool.device.clone(), + pool: Arc::clone(pool), + _init_guard: init_guard, + ptr: mapped.pData as *mut u8, + len, + }), + y_stride, + uv_stride: y_stride, + storage_epoch, + }) +} + +// Samples whose copy to a staging texture can be in flight while an earlier one is mapped. +const READBACK_DEPTH: usize = 2; + pub struct MediaFoundationDecoder { source_reader: IMFSourceReader, d3d11_device: ID3D11Device, @@ -647,16 +918,19 @@ pub struct MediaFoundationDecoder { height: u32, frame_rate_num: u32, frame_rate_den: u32, - readback_free: Vec, readback_queue: VecDeque<(i64, ID3D11Texture2D)>, - cpu_readback_pool: CpuReadbackPool, + staging_pool: Arc, + reclaim_tx: mpsc::Sender, + reclaim_rx: mpsc::Receiver, + readback_stats: NV12ReadbackStats, // How the stream stopped, reported once the queued readbacks are drained. readback_end: Option>, frame_pool: FramePool, plane_converter: Nv12PlaneConverter, capabilities: MFDecoderCapabilities, // COM can unload the decoder DLL, so release every COM field before its apartment shuts down. - _init_guard: MFInitGuard, + _init_guard: Arc, + _com_guard: ComInitGuard, } struct MFInitGuard { @@ -665,11 +939,28 @@ struct MFInitGuard { impl Drop for MFInitGuard { fn drop(&mut self) { - unsafe { - if self.media_foundation_started { + if self.media_foundation_started { + unsafe { let _ = MFShutdown(); } - CoUninitialize(); + } + } +} + +// COM apartments belong to the thread that initialized them, so that lifetime +// stays with the decoder on its creating thread instead of following mapped +// frames onto a render thread. +struct ComInitGuard { + initialized: bool, + init_thread: std::thread::ThreadId, +} + +impl Drop for ComInitGuard { + fn drop(&mut self) { + if self.initialized && std::thread::current().id() == self.init_thread { + unsafe { + CoUninitialize(); + } } } } @@ -680,21 +971,41 @@ impl MediaFoundationDecoder { } unsafe fn new_inner(path: &Path) -> Result { + unsafe { Self::new_inner_with_device(path, || create_d3d11_device()) } + } + + #[cfg(test)] + pub(crate) fn new_plain_warp(path: impl AsRef) -> Result { + unsafe { + Self::new_inner_with_device(path.as_ref(), || { + create_warp_device(false).map_err(|e| format!("plain WARP: {e}")) + }) + } + } + + unsafe fn new_inner_with_device( + path: &Path, + create_device: impl FnOnce() -> Result<(ID3D11Device, ID3D11DeviceContext), String>, + ) -> Result { unsafe { CoInitializeEx(None, COINIT_MULTITHREADED) .ok() .map_err(|e| format!("Failed to initialize COM: {e:?}"))?; } - let mut guard = MFInitGuard { + let com_guard = ComInitGuard { + initialized: true, + init_thread: std::thread::current().id(), + }; + let mut init_guard = MFInitGuard { media_foundation_started: false, }; unsafe { MFStartup(MF_API_VERSION, MFSTARTUP_NOSOCKET) .map_err(|e| format!("Failed to start Media Foundation: {e:?}"))?; } - guard.media_foundation_started = true; + init_guard.media_foundation_started = true; - let (d3d11_device, d3d11_context) = unsafe { create_d3d11_device()? }; + let (d3d11_device, d3d11_context) = create_device()?; let device_manager = unsafe { create_dxgi_device_manager(&d3d11_device)? }; let source_reader = unsafe { create_source_reader(path, &device_manager)? }; @@ -709,10 +1020,12 @@ impl MediaFoundationDecoder { .clone(); let plane_converter = Nv12PlaneConverter::new(&d3d11_device, &d3d11_context)?; - - let readback_free = (0..READBACK_DEPTH) - .map(|_| unsafe { create_staging_texture(&d3d11_device, width, height) }) - .collect::, _>>()?; + let staging_pool = Arc::new(StagingTexturePool::new(d3d11_device.clone(), width, height)); + let (reclaim_tx, reclaim_rx) = mpsc::channel(); + for _ in 0..READBACK_DEPTH { + let texture = unsafe { create_staging_texture(&d3d11_device, width, height)? }; + staging_pool.recycle(texture); + } if width > capabilities.max_width || height > capabilities.max_height { warn!( @@ -741,14 +1054,17 @@ impl MediaFoundationDecoder { height, frame_rate_num, frame_rate_den, - readback_free, readback_queue: VecDeque::new(), - cpu_readback_pool: CpuReadbackPool::default(), + staging_pool, + reclaim_tx, + reclaim_rx, + readback_stats: NV12ReadbackStats::default(), readback_end: None, frame_pool: FramePool::new(), plane_converter, capabilities, - _init_guard: guard, + _init_guard: Arc::new(init_guard), + _com_guard: com_guard, }) } @@ -772,14 +1088,33 @@ impl MediaFoundationDecoder { &self.capabilities } + pub fn readback_stats(&self) -> NV12ReadbackStats { + self.readback_stats + } + /// Reads the next sample back to the CPU. The copy of the sample after it is queued - /// first, so the GPU decodes and copies that one while this one is copied out. + /// first, so the GPU decodes and copies that one while this one is mapped. pub fn read_nv12(&mut self) -> Result, String> { - while self.readback_end.is_none() - && let Some(staging) = self.readback_free.pop() - { + self.drain_reclaimed(); + let mut enqueued = false; + while self.readback_end.is_none() && self.readback_queue.len() < READBACK_DEPTH { + let (staging, reused) = self.staging_pool.acquire()?; + if reused { + self.readback_stats.pool_hits += 1; + } else { + self.readback_stats.pool_misses += 1; + } + match unsafe { self.read_decoder_surface() } { Ok(Some(surface)) => { + let source_box = D3D11_BOX { + left: 0, + top: 0, + front: 0, + right: self.width, + bottom: self.height, + back: 1, + }; unsafe { self.d3d11_context.CopySubresourceRegion( &staging, @@ -789,17 +1124,23 @@ impl MediaFoundationDecoder { 0, &surface.texture, surface.subresource, - None, + Some(&source_box), ); } self.readback_queue.push_back((surface.pts, staging)); + enqueued = true; } end => { - self.readback_free.push(staging); + self.staging_pool.recycle(staging); self.readback_end = Some(end.map(|_| ())); } } } + if enqueued { + unsafe { + self.d3d11_context.Flush(); + } + } let Some((pts, staging)) = self.readback_queue.pop_front() else { return self @@ -808,41 +1149,34 @@ impl MediaFoundationDecoder { .expect("the queue only runs dry once the stream has stopped") .map(|()| None); }; - let data = unsafe { self.map_staging(&staging) }; - self.readback_free.push(staging); + let data = unsafe { self.map_staging(staging) }; Ok(Some(NV12Sample { pts, data: data? })) } - unsafe fn map_staging(&mut self, staging: &ID3D11Texture2D) -> Result { - let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + unsafe fn map_staging(&mut self, staging: ID3D11Texture2D) -> Result { unsafe { - self.d3d11_context - .Map(staging, 0, D3D11_MAP_READ, 0, Some(&mut mapped)) - .map_err(|e| format!("Map staging texture failed: {e:?}"))?; + map_nv12_staging( + &self.d3d11_context, + staging, + self.height, + &self.staging_pool, + &self.reclaim_tx, + &mut self.readback_stats, + Arc::clone(&self._init_guard), + ) } + } - let y_stride = mapped.RowPitch; - let y_height = self.height; - let uv_height = self.height / 2; - - let y_size = (y_stride * y_height) as usize; - let uv_size = (y_stride * uv_height) as usize; - let total_size = y_size + uv_size; - - let mut data = self.cpu_readback_pool.acquire(total_size); - // SAFETY: the mapped NV12 resource contains `total_size` bytes and the pool reserves that - // much destination capacity. The copy initializes every byte before setting the length. - unsafe { - std::ptr::copy_nonoverlapping(mapped.pData as *const u8, data.as_mut_ptr(), total_size); - data.set_len(total_size); - self.d3d11_context.Unmap(staging, 0); + fn drain_reclaimed(&mut self) { + while let Ok(mut reclaim) = self.reclaim_rx.try_recv() { + let Some(texture) = reclaim.take() else { + continue; + }; + unsafe { + self.d3d11_context.Unmap(&texture, 0); + } + self.staging_pool.recycle(texture); } - - Ok(NV12Data { - data: Arc::new(data), - y_stride, - uv_stride: y_stride, - }) } pub fn read_sample(&mut self) -> Result, String> { @@ -857,10 +1191,6 @@ impl MediaFoundationDecoder { } } - pub fn recycle_nv12_buffer(&mut self, buffer: Arc>) { - self.cpu_readback_pool.recycle(buffer); - } - unsafe fn read_decoder_surface(&mut self) -> Result, String> { let mut stream_index = 0u32; let mut flags = 0u32; @@ -993,13 +1323,21 @@ impl MediaFoundationDecoder { .map_err(|e| format!("Seek failed: {e:?}"))?; } - self.readback_free - .extend(self.readback_queue.drain(..).map(|(_, staging)| staging)); + self.drain_reclaimed(); + for (_, staging) in self.readback_queue.drain(..) { + self.staging_pool.recycle(staging); + } self.readback_end = None; Ok(()) } } +impl Drop for MediaFoundationDecoder { + fn drop(&mut self) { + self.drain_reclaimed(); + } +} + unsafe fn create_d3d11_device() -> Result<(ID3D11Device, ID3D11DeviceContext), String> { use windows::Win32::Graphics::Direct3D::D3D_DRIVER_TYPE_UNKNOWN; @@ -1067,10 +1405,36 @@ unsafe fn create_d3d11_device() -> Result<(ID3D11Device, ID3D11DeviceContext), S warn!("No physical hardware DXGI adapter available, falling back to WARP"); } + match unsafe { create_warp_device(true) } { + Ok(device) => { + warn!("Using WARP software rasterizer for D3D11 - hardware GPU unavailable"); + Ok(device) + } + Err(error) => Err(format!("{last_error}; {error}")), + } +} + +unsafe fn create_warp_device( + video_support: bool, +) -> Result<(ID3D11Device, ID3D11DeviceContext), String> { + use windows::Win32::Graphics::Direct3D::D3D_DRIVER_TYPE_WARP; + + let mut flags = D3D11_CREATE_DEVICE_BGRA_SUPPORT; + if video_support { + flags |= D3D11_CREATE_DEVICE_VIDEO_SUPPORT; + } + + let feature_levels = [ + windows::Win32::Graphics::Direct3D::D3D_FEATURE_LEVEL_11_1, + windows::Win32::Graphics::Direct3D::D3D_FEATURE_LEVEL_11_0, + windows::Win32::Graphics::Direct3D::D3D_FEATURE_LEVEL_10_1, + windows::Win32::Graphics::Direct3D::D3D_FEATURE_LEVEL_10_0, + ]; + let mut device: Option = None; let mut context: Option = None; - let result = unsafe { + unsafe { D3D11CreateDevice( None, D3D_DRIVER_TYPE_WARP, @@ -1082,30 +1446,22 @@ unsafe fn create_d3d11_device() -> Result<(ID3D11Device, ID3D11DeviceContext), S None, Some(&mut context), ) - }; - - match result { - Ok(()) => { - if let (Some(device), Some(context)) = (device, context) { - warn!("Using WARP software rasterizer for D3D11 - hardware GPU unavailable"); + } + .map_err(|error| format!("D3D11CreateDevice (WARP) failed: {error:?}"))?; - let multithread: windows::Win32::Graphics::Direct3D11::ID3D11Multithread = device - .cast() - .map_err(|e| format!("Failed to get ID3D11Multithread: {e:?}"))?; - unsafe { - let _ = multithread.SetMultithreadProtected(true); - } + let (device, context) = match (device, context) { + (Some(device), Some(context)) => (device, context), + _ => return Err("D3D11CreateDevice (WARP) returned a null device/context".to_string()), + }; - return Ok((device, context)); - } - Err(format!( - "{last_error}; D3D11CreateDevice (WARP) returned null device/context" - )) - } - Err(e) => Err(format!( - "{last_error}; D3D11CreateDevice (WARP) failed: {e:?}" - )), + let multithread: windows::Win32::Graphics::Direct3D11::ID3D11Multithread = device + .cast() + .map_err(|e| format!("Failed to get ID3D11Multithread: {e:?}"))?; + unsafe { + let _ = multithread.SetMultithreadProtected(true); } + + Ok((device, context)) } unsafe fn create_dxgi_device_manager( @@ -1241,84 +1597,239 @@ unsafe impl Send for MediaFoundationDecoder {} #[cfg(test)] mod tests { use super::*; + use std::sync::mpsc::TryRecvError; + use std::sync::{Mutex, MutexGuard}; - #[test] - fn cpu_readback_pool_reuses_recycled_buffers() { - let mut pool = CpuReadbackPool::default(); - let mut buffer = pool.acquire(16); - buffer.resize(16, 7); - let pointer = buffer.as_ptr(); - pool.recycle(Arc::new(buffer)); + const FIXTURE_1080P: &str = "h264-1080p.mp4"; + static MF_TEST_LOCK: Mutex<()> = Mutex::new(()); - let reused = pool.acquire(8); - assert_eq!(reused.as_ptr(), pointer); - assert!(reused.is_empty()); + fn serial_mf_test() -> MutexGuard<'static, ()> { + MF_TEST_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) } - #[test] - fn cpu_readback_pool_handles_changed_buffer_sizes() { - let mut pool = CpuReadbackPool::default(); - let mut buffer = pool.acquire(16); - buffer.resize(16, 7); - pool.recycle(Arc::new(buffer)); + fn fixture(name: &str) -> std::path::PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("tests/fixtures") + .join(name) + } - let mut grown = pool.acquire(64); - assert!(grown.capacity() >= 64); - assert!(grown.is_empty()); - grown.resize(64, 9); - let pointer = grown.as_ptr(); - pool.recycle(Arc::new(grown)); + fn skip_or_fail(required: bool, reason: &str) { + if required { + panic!("{reason}"); + } + eprintln!("MediaFoundation test skipped: {reason}"); + } - let reduced = pool.acquire(8); - assert_eq!(reduced.as_ptr(), pointer); - assert!(reduced.is_empty()); + fn skip_or_fail_d3d11(reason: &str) { + skip_or_fail( + std::env::var_os("CAP_REQUIRE_D3D11_TESTS").is_some(), + reason, + ); } - #[test] - fn cpu_readback_pool_waits_for_shared_buffers_to_be_released() { - let mut pool = CpuReadbackPool::default(); - let buffer = Arc::new(vec![7; 16]); - let pointer = buffer.as_ptr(); - let renderer_reference = Arc::clone(&buffer); - pool.recycle(buffer); + fn skip_or_fail_mf(reason: &str) { + skip_or_fail(std::env::var_os("CAP_REQUIRE_MF_TESTS").is_some(), reason); + } + + fn test_device() -> Option<(ID3D11Device, ID3D11DeviceContext)> { + if std::env::var_os("CAP_FORCE_WARP_TESTS").is_none() + && let Ok(device) = unsafe { create_d3d11_device() } + { + return Some(device); + } + + // Hosted Windows runners expose no video-capable adapter. The staging + // tests only need a D3D11 context, so fall back to plain WARP. + match unsafe { create_warp_device(false) } { + Ok(device) => Some(device), + Err(error) => { + skip_or_fail_d3d11(&format!("no D3D11 test device available: {error}")); + None + } + } + } + + fn test_decoder(path: &Path) -> Option { + let result = if std::env::var_os("CAP_FORCE_WARP_TESTS").is_some() { + MediaFoundationDecoder::new_plain_warp(path) + } else { + MediaFoundationDecoder::new(path) + }; + match result { + Ok(decoder) => Some(decoder), + Err(error) => { + skip_or_fail_mf(&format!( + "no Media Foundation H.264 decoder available: {error}" + )); + None + } + } + } + + fn create_cpu_readwrite_staging( + device: &ID3D11Device, + width: u32, + height: u32, + ) -> ID3D11Texture2D { + use windows::Win32::Graphics::Direct3D11::D3D11_CPU_ACCESS_WRITE; + + let desc = D3D11_TEXTURE2D_DESC { + Width: width, + Height: height, + MipLevels: 1, + ArraySize: 1, + Format: DXGI_FORMAT_NV12, + SampleDesc: DXGI_SAMPLE_DESC { + Count: 1, + Quality: 0, + }, + Usage: D3D11_USAGE_STAGING, + BindFlags: 0, + CPUAccessFlags: (D3D11_CPU_ACCESS_READ.0 | D3D11_CPU_ACCESS_WRITE.0) as u32, + MiscFlags: 0, + }; + unsafe { + let mut texture = None; + device + .CreateTexture2D(&desc, None, Some(&mut texture)) + .unwrap(); + texture.unwrap() + } + } + + unsafe fn blocking_map_copy( + context: &ID3D11DeviceContext, + staging: &ID3D11Texture2D, + height: u32, + ) -> (Vec, usize) { + let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + unsafe { + context + .Map(staging, 0, D3D11_MAP_READ, 0, Some(&mut mapped)) + .unwrap(); + let y_stride = mapped.RowPitch as usize; + let len = y_stride * height as usize + y_stride * height.div_ceil(2) as usize; + let data = std::slice::from_raw_parts(mapped.pData as *const u8, len).to_vec(); + context.Unmap(staging, 0); + (data, y_stride) + } + } + + fn assert_nv12_visible_rows_equal( + actual: &[u8], + actual_pitch: usize, + expected: &[u8], + expected_pitch: usize, + width: u32, + height: u32, + ) { + assert_eq!(actual_pitch, expected_pitch); + let luma_rows = height as usize; + let chroma_rows = height.div_ceil(2) as usize; + let visible = width as usize; + assert_eq!(actual.len(), actual_pitch * (luma_rows + chroma_rows)); + assert_eq!(expected.len(), expected_pitch * (luma_rows + chroma_rows)); + + for row in 0..luma_rows { + let start = row * actual_pitch; + assert_eq!( + &actual[start..start + visible], + &expected[start..start + visible], + "luma row {row} differs" + ); + } + let chroma_start = actual_pitch * luma_rows; + for row in 0..chroma_rows { + let start = chroma_start + row * actual_pitch; + assert_eq!( + &actual[start..start + visible], + &expected[start..start + visible], + "chroma row {row} differs" + ); + } + } - let separate = pool.acquire(16); - assert_ne!(separate.as_ptr(), pointer); - drop(separate); - drop(renderer_reference); + fn assert_nv12_samples_equal( + actual: &(i64, usize, Vec), + expected: &(i64, usize, Vec), + width: u32, + height: u32, + ) { + assert_eq!(actual.0, expected.0); + assert_nv12_visible_rows_equal(&actual.2, actual.1, &expected.2, expected.1, width, height); + } - let reused = pool.acquire(8); - assert_eq!(reused.as_ptr(), pointer); - assert!(reused.is_empty()); + fn recycle_reclaimed( + context: &ID3D11DeviceContext, + pool: &StagingTexturePool, + rx: &mpsc::Receiver, + ) { + while let Ok(mut reclaim) = rx.try_recv() { + let Some(texture) = reclaim.take() else { + continue; + }; + unsafe { + context.Unmap(&texture, 0); + } + pool.recycle(texture); + } } #[test] - #[ignore = "requires a Windows D3D11 device"] fn repeatedly_releases_decoder_objects_before_com_shutdown() { - let path = - Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/h264-decoder-lifecycle.mp4"); - for _ in 0..20 { - let mut decoder = MediaFoundationDecoder::new(&path).unwrap(); + let _mf = serial_mf_test(); + if std::env::var_os("CAP_FORCE_WARP_TESTS").is_some() { + eprintln!( + "MediaFoundation lifecycle test skipped: WARP removes devices after repeated re-creation" + ); + return; + } + let path = fixture("h264-decoder-lifecycle.mp4"); + let decoders = 20; + for _ in 0..decoders { + let Some(mut decoder) = test_decoder(&path) else { + return; + }; for _ in 0..5 { assert!(decoder.read_sample().unwrap().is_some()); } drop(decoder); } + println!("MediaFoundation lifecycle: {decoders} decoders created and released"); } #[test] - #[ignore = "requires a Windows D3D11 device"] fn pipelined_readback_returns_every_sample_in_order() { - let path = - Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/h264-decoder-lifecycle.mp4"); + let _mf = serial_mf_test(); + let path = fixture(FIXTURE_1080P); // This file, like the reference recording, ends with a stream error rather than an // end-of-stream flag, so both readers stop at whichever comes first. let mut expected = Vec::new(); - let mut decoder = MediaFoundationDecoder::new(&path).unwrap(); - let staging = decoder.readback_free[0].clone(); + let Some(mut decoder) = test_decoder(&path) else { + return; + }; + let video_width = decoder.width; + let video_height = decoder.height; + let mut surface_height = 0; let expected_end = loop { + let (staging, _) = decoder.staging_pool.acquire().unwrap(); match unsafe { decoder.read_decoder_surface() } { Ok(Some(surface)) => unsafe { + let mut desc = D3D11_TEXTURE2D_DESC::default(); + surface.texture.GetDesc(&mut desc); + surface_height = desc.Height; + // Keep the copy box identical to read_nv12 so this comparison + // isolates the CPU readback path from the surface clipping. + let source_box = D3D11_BOX { + left: 0, + top: 0, + front: 0, + right: video_width, + bottom: video_height, + back: 1, + }; decoder.d3d11_context.CopySubresourceRegion( &staging, 0, @@ -1327,36 +1838,486 @@ mod tests { 0, &surface.texture, surface.subresource, - None, + Some(&source_box), ); - expected.push((surface.pts, decoder.map_staging(&staging).unwrap().data)); + let (data, pitch) = + blocking_map_copy(&decoder.d3d11_context, &staging, video_height); + expected.push((surface.pts, pitch, data)); + decoder.staging_pool.recycle(staging); }, - end => break end.is_err(), + end => { + decoder.staging_pool.recycle(staging); + break end.is_err(); + } } }; drop(decoder); - let mut decoder = MediaFoundationDecoder::new(&path).unwrap(); + let Some(mut decoder) = test_decoder(&path) else { + return; + }; let mut read = Vec::new(); let end = loop { match decoder.read_nv12() { - Ok(Some(sample)) => read.push((sample.pts, sample.data.data)), + Ok(Some(sample)) => read.push(( + sample.pts, + sample.data.y_stride as usize, + sample.data.as_slice().to_vec(), + )), end => break end.is_err(), } }; + assert_eq!((video_width, video_height), (1920, 1080)); + assert_eq!( + surface_height, 1088, + "the decoder surface must stay macroblock-aligned for this test" + ); assert!(expected.len() > READBACK_DEPTH); - assert!(read == expected); + assert_eq!(read.len(), expected.len()); + for (actual, expected) in read.iter().zip(&expected) { + assert_nv12_samples_equal(actual, expected, video_width, video_height); + } assert_eq!(end, expected_end); + let stats = decoder.readback_stats(); + assert_eq!(stats.frames, read.len() as u64); + assert!(stats.pool_hits + stats.pool_misses >= stats.frames); + println!( + "MediaFoundation readback parity: {} frames, {}x{}, {}-row decoder surface", + read.len(), + video_width, + video_height, + surface_height + ); + let first = |decoder: &mut MediaFoundationDecoder| { let sample = decoder.read_nv12().unwrap().unwrap(); - (sample.pts, sample.data.data) + ( + sample.pts, + sample.data.y_stride as usize, + sample.data.as_slice().to_vec(), + ) }; decoder.seek(0).unwrap(); - assert!(first(&mut decoder) == expected[0]); + assert_nv12_samples_equal( + &first(&mut decoder), + &expected[0], + video_width, + video_height, + ); decoder.read_nv12().unwrap(); decoder.seek(0).unwrap(); - assert!(first(&mut decoder) == expected[0]); + assert_nv12_samples_equal( + &first(&mut decoder), + &expected[0], + video_width, + video_height, + ); + } + + #[test] + fn software_decoded_1080p_frame_matches_blocking_readback() { + use windows::Win32::Graphics::Direct3D11::D3D11_MAP_WRITE; + + const WIDTH: u32 = 1920; + const HEIGHT: u32 = 1080; + const SURFACE_HEIGHT: u32 = 1088; + + let _mf = serial_mf_test(); + let path = fixture(FIXTURE_1080P); + let mut decoder = crate::FFmpegDecoder::new(&path, None).unwrap(); + let frame = decoder + .frames() + .next() + .expect("fixture has no frames") + .unwrap(); + assert_eq!(frame.format(), ::ffmpeg::format::Pixel::YUV420P); + + let Some((device, context)) = test_device() else { + return; + }; + let source = create_cpu_readwrite_staging(&device, WIDTH, SURFACE_HEIGHT); + let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + unsafe { + context + .Map(&source, 0, D3D11_MAP_WRITE, 0, Some(&mut mapped)) + .unwrap(); + let pitch = mapped.RowPitch as usize; + let total = pitch * (SURFACE_HEIGHT + SURFACE_HEIGHT.div_ceil(2)) as usize; + let bytes = std::slice::from_raw_parts_mut(mapped.pData as *mut u8, total); + + for row in 0..SURFACE_HEIGHT as usize { + let destination = &mut bytes[row * pitch..row * pitch + WIDTH as usize]; + if row < HEIGHT as usize { + let stride = frame.stride(0); + let source = &frame.data(0)[row * stride..row * stride + WIDTH as usize]; + destination.copy_from_slice(source); + } else { + destination.fill(0xEE); + } + } + + let uv_start = pitch * SURFACE_HEIGHT as usize; + for row in 0..SURFACE_HEIGHT.div_ceil(2) as usize { + let start = uv_start + row * pitch; + let destination = &mut bytes[start..start + WIDTH as usize]; + if row < HEIGHT.div_ceil(2) as usize { + let u_stride = frame.stride(1); + let v_stride = frame.stride(2); + let u = &frame.data(1)[row * u_stride..row * u_stride + WIDTH as usize / 2]; + let v = &frame.data(2)[row * v_stride..row * v_stride + WIDTH as usize / 2]; + for (x, (&u_byte, &v_byte)) in u.iter().zip(v).enumerate() { + destination[2 * x] = u_byte; + destination[2 * x + 1] = v_byte; + } + } else { + destination.fill(0xDD); + } + } + context.Unmap(&source, 0); + context.Flush(); + } + + let pool = Arc::new(StagingTexturePool::new(device.clone(), WIDTH, HEIGHT)); + let (old_staging, _) = pool.acquire().unwrap(); + let (new_staging, _) = pool.acquire().unwrap(); + let source_box = D3D11_BOX { + left: 0, + top: 0, + front: 0, + right: WIDTH, + bottom: HEIGHT, + back: 1, + }; + unsafe { + for destination in [&old_staging, &new_staging] { + context.CopySubresourceRegion( + destination, + 0, + 0, + 0, + 0, + &source, + 0, + Some(&source_box), + ); + } + context.Flush(); + } + + let (old, old_pitch) = unsafe { blocking_map_copy(&context, &old_staging, HEIGHT) }; + pool.recycle(old_staging); + + let (reclaim_tx, reclaim_rx) = mpsc::channel(); + let mut stats = NV12ReadbackStats::default(); + let guard = Arc::new(MFInitGuard { + media_foundation_started: false, + }); + let new = unsafe { + map_nv12_staging( + &context, + new_staging, + HEIGHT, + &pool, + &reclaim_tx, + &mut stats, + guard, + ) + .unwrap() + }; + assert_nv12_visible_rows_equal( + new.as_slice(), + new.y_stride as usize, + &old, + old_pitch, + WIDTH, + HEIGHT, + ); + println!( + "MediaFoundation software-decode parity: 1 frame, {WIDTH}x{HEIGHT}, {SURFACE_HEIGHT}-row source" + ); + drop(new); + recycle_reclaimed(&context, &pool, &reclaim_rx); + } + + #[test] + fn mapped_frame_outlives_decoder() { + let _mf = serial_mf_test(); + let path = fixture(FIXTURE_1080P); + let Some(mut decoder) = test_decoder(&path) else { + return; + }; + let frame = decoder.read_nv12().unwrap().unwrap(); + drop(decoder); + + let byte = std::thread::spawn(move || { + let byte = frame.data.as_slice()[0]; + drop(frame); + byte + }) + .join() + .unwrap(); + std::hint::black_box(byte); + println!("MediaFoundation mapped frame outlived its decoder"); + } + + #[test] + fn dropped_mapped_frame_is_reclaimed_when_the_decoder_drains() { + let _mf = serial_mf_test(); + let path = fixture(FIXTURE_1080P); + let Some(mut decoder) = test_decoder(&path) else { + return; + }; + let frame = decoder.read_nv12().unwrap().unwrap(); + let free_before = decoder.staging_pool.free.lock().unwrap().textures.len(); + drop(frame); + let free_after_drop = decoder.staging_pool.free.lock().unwrap().textures.len(); + assert_eq!(free_after_drop, free_before); + + decoder.drain_reclaimed(); + let free_after_drain = decoder.staging_pool.free.lock().unwrap().textures.len(); + assert_eq!(free_after_drain, free_before + 1); + assert!(matches!( + decoder.reclaim_rx.try_recv(), + Err(TryRecvError::Empty) + )); + println!("MediaFoundation mapped frame reclaimed when the decoder drained it"); + } + + #[test] + fn staging_pool_reuses_recycled_textures() { + let Some((device, _)) = test_device() else { + return; + }; + let pool = StagingTexturePool::new(device, 64, 64); + + let (first, first_hit) = pool.acquire().unwrap(); + assert!(!first_hit); + pool.recycle(first); + + let (second, second_hit) = pool.acquire().unwrap(); + assert!(second_hit); + pool.recycle(second); + println!("MediaFoundation staging pool reused a recycled texture"); + } + + #[test] + fn aligned_decoder_surface_copies_only_the_visible_rows() { + use windows::Win32::Graphics::Direct3D11::D3D11_MAP_WRITE; + + const WIDTH: u32 = 1920; + const HEIGHT: u32 = 1080; + const SURFACE_HEIGHT: u32 = 1088; + + let Some((device, context)) = test_device() else { + return; + }; + let source = create_cpu_readwrite_staging(&device, WIDTH, SURFACE_HEIGHT); + let staging = unsafe { create_staging_texture(&device, WIDTH, HEIGHT).unwrap() }; + + unsafe { + let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + context + .Map(&source, 0, D3D11_MAP_WRITE, 0, Some(&mut mapped)) + .unwrap(); + let pitch = mapped.RowPitch as usize; + let y_len = pitch * SURFACE_HEIGHT as usize; + let uv_len = pitch * SURFACE_HEIGHT.div_ceil(2) as usize; + let bytes = std::slice::from_raw_parts_mut(mapped.pData as *mut u8, y_len + uv_len); + for row in 0..SURFACE_HEIGHT as usize { + let value = if row < HEIGHT as usize { + row as u8 + } else { + 0xEE + }; + bytes[row * pitch..(row + 1) * pitch].fill(value); + } + for row in 0..SURFACE_HEIGHT.div_ceil(2) as usize { + let value = if row < HEIGHT.div_ceil(2) as usize { + 0x80 | (row as u8 & 0x7F) + } else { + 0xDD + }; + let start = y_len + row * pitch; + bytes[start..start + pitch].fill(value); + } + context.Unmap(&source, 0); + + let source_box = D3D11_BOX { + left: 0, + top: 0, + front: 0, + right: WIDTH, + bottom: HEIGHT, + back: 1, + }; + context.CopySubresourceRegion(&staging, 0, 0, 0, 0, &source, 0, Some(&source_box)); + context.Flush(); + + context + .Map(&staging, 0, D3D11_MAP_READ, 0, Some(&mut mapped)) + .unwrap(); + let staging_pitch = mapped.RowPitch as usize; + let staging_y_len = staging_pitch * HEIGHT as usize; + let staging_uv_len = staging_pitch * HEIGHT.div_ceil(2) as usize; + let visible = std::slice::from_raw_parts( + mapped.pData as *const u8, + staging_y_len + staging_uv_len, + ); + for row in 0..HEIGHT as usize { + assert!( + visible[row * staging_pitch..row * staging_pitch + WIDTH as usize] + .iter() + .all(|&byte| byte == row as u8), + "luma row {row} was not copied from the visible region" + ); + } + for row in 0..HEIGHT.div_ceil(2) as usize { + let start = staging_y_len + row * staging_pitch; + assert!( + visible[start..start + WIDTH as usize] + .iter() + .all(|&byte| byte == 0x80 | (row as u8 & 0x7F)), + "chroma row {row} was not copied from the visible region" + ); + } + context.Unmap(&staging, 0); + } + println!("MediaFoundation 1088-row copy: only visible 1080 luma and chroma rows"); + } + + #[test] + #[ignore = "performance probe: baseline blocking Map plus a fresh buffer per frame"] + fn nv12_4k_readback_baseline() { + const WIDTH: u32 = 3840; + const HEIGHT: u32 = 2160; + const ITERS: u32 = 60; + + let (device, context) = unsafe { create_d3d11_device().unwrap() }; + let source = create_internal_texture(&device, WIDTH, HEIGHT, DXGI_FORMAT_NV12, 0).unwrap(); + let staging = [ + unsafe { create_staging_texture(&device, WIDTH, HEIGHT).unwrap() }, + unsafe { create_staging_texture(&device, WIDTH, HEIGHT).unwrap() }, + ]; + + unsafe { + context.CopySubresourceRegion(&staging[0], 0, 0, 0, 0, &source.texture, 0, None); + context.Flush(); + } + + let wall_start = Instant::now(); + let mut map_wait = Duration::ZERO; + let mut copy_time = Duration::ZERO; + let mut bytes = 0usize; + for iteration in 0..ITERS { + let current = (iteration % 2) as usize; + let next = ((iteration + 1) % 2) as usize; + unsafe { + context.CopySubresourceRegion(&staging[next], 0, 0, 0, 0, &source.texture, 0, None); + context.Flush(); + + let map_start = Instant::now(); + let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + context + .Map(&staging[current], 0, D3D11_MAP_READ, 0, Some(&mut mapped)) + .unwrap(); + map_wait += map_start.elapsed(); + + let y_stride = mapped.RowPitch as usize; + let total_size = + y_stride * HEIGHT as usize + y_stride * HEIGHT.div_ceil(2) as usize; + let copy_start = Instant::now(); + let mut data = vec![0u8; total_size]; + std::ptr::copy_nonoverlapping( + mapped.pData as *const u8, + data.as_mut_ptr(), + total_size, + ); + context.Unmap(&staging[current], 0); + copy_time += copy_start.elapsed(); + bytes += total_size; + std::hint::black_box(data.len()); + } + } + let wall = wall_start.elapsed(); + + let frames = ITERS as f64; + let readback_mib = bytes as f64 / (1024.0 * 1024.0); + println!( + "4K NV12 MF readback baseline (blocking map, fresh Vec): {:.2} ms/frame wall ({:.1} MiB/s), map {:.2} ms, alloc+copy {:.2} ms", + wall.as_secs_f64() * 1000.0 / frames, + readback_mib / wall.as_secs_f64(), + map_wait.as_secs_f64() * 1000.0 / frames, + copy_time.as_secs_f64() * 1000.0 / frames, + ); + } + + #[test] + #[ignore = "performance probe: requires a Windows D3D11 device"] + fn nv12_4k_readback_benchmark() { + const WIDTH: u32 = 3840; + const HEIGHT: u32 = 2160; + const ITERS: u32 = 60; + + let (device, context) = unsafe { create_d3d11_device().unwrap() }; + let source = create_internal_texture(&device, WIDTH, HEIGHT, DXGI_FORMAT_NV12, 0).unwrap(); + let staging_pool = Arc::new(StagingTexturePool::new(device.clone(), WIDTH, HEIGHT)); + let mut stats = NV12ReadbackStats::default(); + + let (mut staging, _) = staging_pool.acquire().unwrap(); + unsafe { + context.CopySubresourceRegion(&staging, 0, 0, 0, 0, &source.texture, 0, None); + context.Flush(); + } + + let guard = Arc::new(MFInitGuard { + media_foundation_started: false, + }); + let (reclaim_tx, reclaim_rx) = mpsc::channel(); + + let wall_start = Instant::now(); + for _ in 0..ITERS { + let (next, reused) = staging_pool.acquire().unwrap(); + if reused { + stats.pool_hits += 1; + } else { + stats.pool_misses += 1; + } + unsafe { + context.CopySubresourceRegion(&next, 0, 0, 0, 0, &source.texture, 0, None); + context.Flush(); + let data = map_nv12_staging( + &context, + staging, + HEIGHT, + &staging_pool, + &reclaim_tx, + &mut stats, + Arc::clone(&guard), + ) + .unwrap(); + std::hint::black_box(data.as_slice().len()); + } + recycle_reclaimed(&context, &staging_pool, &reclaim_rx); + staging = next; + } + let wall = wall_start.elapsed(); + + let frames = stats.frames as f64; + let readback_mib = stats.bytes as f64 / (1024.0 * 1024.0); + println!( + "4K NV12 MF readback (2-deep, zero-copy): {:.2} ms/frame wall ({:.1} MiB/s), map {:.2} ms, immediate {}/{}, blocked {}, staging hits {}", + wall.as_secs_f64() * 1000.0 / frames, + readback_mib / wall.as_secs_f64(), + stats.map_wait.as_secs_f64() * 1000.0 / frames, + stats.map_immediate, + stats.frames, + stats.map_blocked, + stats.pool_hits, + ); + assert_eq!(stats.frames, ITERS as u64); + assert!(stats.pool_hits + stats.pool_misses >= stats.frames); + assert!(stats.pool_hits > 0); } fn normalized(path: &str) -> String { diff --git a/crates/video-decode/tests/fixtures/h264-1080p.mp4 b/crates/video-decode/tests/fixtures/h264-1080p.mp4 new file mode 100644 index 00000000000..b5f4a943fcc Binary files /dev/null and b/crates/video-decode/tests/fixtures/h264-1080p.mp4 differ