diff --git a/crates/rendering/src/decoder/media_foundation.rs b/crates/rendering/src/decoder/media_foundation.rs index 6d77001006d..19e62f6d666 100644 --- a/crates/rendering/src/decoder/media_foundation.rs +++ b/crates/rendering/src/decoder/media_foundation.rs @@ -3,12 +3,11 @@ use std::{ collections::BTreeMap, path::PathBuf, - sync::{Arc, mpsc}, + sync::mpsc, time::{Duration, Instant}, }; use tokio::sync::oneshot; use tracing::{info, warn}; -use windows::Win32::{Foundation::HANDLE, Graphics::Direct3D11::ID3D11Texture2D}; use super::{DecodedFrame, DecoderInitResult, DecoderType, FRAME_CACHE_SIZE, VideoDecoderMessage}; @@ -16,7 +15,6 @@ const MAX_FRAME_CACHE_BYTES: usize = 128 * 1024 * 1024; struct DecoderHealthMonitor { consecutive_errors: u32, - consecutive_texture_read_failures: u32, total_frames_decoded: u64, total_errors: u64, last_successful_decode: Instant, @@ -31,7 +29,6 @@ impl DecoderHealthMonitor { fn new() -> Self { Self { consecutive_errors: 0, - consecutive_texture_read_failures: 0, total_frames_decoded: 0, total_errors: 0, last_successful_decode: Instant::now(), @@ -49,7 +46,6 @@ impl DecoderHealthMonitor { fn record_success(&mut self, decode_time: Duration) { self.consecutive_errors = 0; - self.consecutive_texture_read_failures = 0; self.total_frames_decoded += 1; self.last_successful_decode = Instant::now(); @@ -67,18 +63,11 @@ impl DecoderHealthMonitor { self.total_errors += 1; } - fn record_texture_read_failure(&mut self) { - self.consecutive_texture_read_failures += 1; - } - fn is_healthy(&self) -> bool { const MAX_CONSECUTIVE_ERRORS: u32 = 10; - const MAX_CONSECUTIVE_TEXTURE_FAILURES: u32 = 5; const MAX_TIME_SINCE_SUCCESS: Duration = Duration::from_secs(5); - if self.consecutive_errors >= MAX_CONSECUTIVE_ERRORS - || self.consecutive_texture_read_failures >= MAX_CONSECUTIVE_TEXTURE_FAILURES - { + if self.consecutive_errors >= MAX_CONSECUTIVE_ERRORS { return false; } @@ -122,63 +111,6 @@ impl DecoderHealthMonitor { } } -#[derive(Clone)] -struct CachedFrame { - number: u32, - _texture: Option, - _shared_handle: Option, - _y_handle: Option, - _uv_handle: Option, - nv12_data: Option>, - width: u32, - height: u32, -} - -impl CachedFrame { - fn estimated_bytes(&self) -> usize { - let texture_bytes = self._texture.as_ref().map_or(0, |_| { - (self.width as usize) - .saturating_mul(self.height as usize) - .saturating_mul(3) - }); - texture_bytes.saturating_add(self.nv12_data.as_ref().map_or(0, |data| data.data.len())) - } - - fn to_decoded_frame(&self) -> DecodedFrame { - let null_ptr = std::ptr::null_mut(); - let y_handle = self._y_handle.filter(|h| h.0 != null_ptr); - let uv_handle = self._uv_handle.filter(|h| h.0 != null_ptr); - if let (Some(texture), Some(y_handle), Some(uv_handle)) = - (&self._texture, y_handle, uv_handle) - { - return DecodedFrame::new_nv12_with_d3d11_texture_and_yuv_handles( - self.width, - self.height, - texture.clone(), - self._shared_handle, - Some(y_handle), - Some(uv_handle), - ); - } - - if let Some(nv12_data) = &self.nv12_data { - DecodedFrame::new_nv12_with_arc( - Arc::clone(&nv12_data.data), - self.width, - self.height, - nv12_data.y_stride, - nv12_data.uv_stride, - ) - } else { - warn!( - "CachedFrame has no CPU data, creating black frame (D3D11 zero-copy not implemented)" - ); - let black_data = vec![0u8; (self.width * self.height * 3 / 2) as usize]; - DecodedFrame::new_nv12(black_data, self.width, self.height, self.width, self.width) - } - } -} - pub struct MFDecoder; impl MFDecoder { @@ -249,7 +181,8 @@ impl MFDecoder { 30 }; let mut requests = FrameRequests::new(readahead_frames); - let answer = |sender: oneshot::Sender, frame: Option| { + let answer = |sender: oneshot::Sender, + frame: Option| { let _ = sender.send(frame.map_or_else( || { DecodedFrame::new( @@ -258,7 +191,16 @@ impl MFDecoder { video_height, ) }, - |frame| frame.to_decoded_frame().with_matrix(matrix), + |data| { + DecodedFrame::new_nv12_with_arc( + data.data, + video_width, + video_height, + data.y_stride, + data.uv_stride, + ) + .with_matrix(matrix) + }, )); }; @@ -290,7 +232,6 @@ impl MFDecoder { warn!( name = name, consecutive_errors = health.consecutive_errors, - texture_failures = health.consecutive_texture_read_failures, total_decoded = health.total_frames_decoded, "MediaFoundation decoder unhealthy, performance may degrade" ); @@ -314,24 +255,22 @@ impl MFDecoder { }, Step::Decode => { let decode_start = Instant::now(); - match decoder.read_sample() { - Ok(Some(mf_frame)) => { - let cached = cache_sample( - &mut decoder, - mf_frame, - decode_start.elapsed(), - &mut health, - fps, - ); - let (number, bytes) = (cached.number, cached.estimated_bytes()); - requests.decoded(number, cached, bytes) + match decoder.read_nv12() { + Ok(Some(sample)) => { + health.record_success(decode_start.elapsed()); + let bytes = sample.data.data.len(); + requests.decoded( + pts_100ns_to_frame(sample.pts, fps), + sample.data, + bytes, + ) } Ok(None) => requests.stopped(), Err(e) => { health.record_error(); warn!( consecutive_errors = health.consecutive_errors, - "MediaFoundation read_sample error: {e}" + "MediaFoundation read error: {e}" ); requests.stopped() } @@ -359,58 +298,6 @@ fn pts_100ns_to_frame(pts_100ns: i64, fps: u32) -> u32 { super::pts_to_frame(pts_100ns, ::ffmpeg::Rational::new(1, 10_000_000), fps) } -fn cache_sample( - decoder: &mut cap_video_decode::MediaFoundationDecoder, - mf_frame: cap_video_decode::MFDecodedFrame, - decode_time: Duration, - health: &mut DecoderHealthMonitor, - fps: u32, -) -> CachedFrame { - let frame_number = pts_100ns_to_frame(mf_frame.pts, fps); - - let has_valid_zero_copy_handles = { - let null_ptr = std::ptr::null_mut(); - mf_frame.textures.y.handle.0 != null_ptr && mf_frame.textures.uv.handle.0 != null_ptr - }; - - let nv12_data = if has_valid_zero_copy_handles { - health.record_success(decode_time); - None - } else { - match decoder.read_texture_to_cpu( - &mf_frame.textures.nv12.texture, - mf_frame.width, - mf_frame.height, - ) { - Ok(data) => { - health.record_success(decode_time); - Some(Arc::new(data)) - } - Err(e) => { - health.record_texture_read_failure(); - warn!("Failed to read texture to CPU for frame {frame_number}: {e}"); - None - } - } - }; - - let cached = CachedFrame { - number: frame_number, - _texture: has_valid_zero_copy_handles.then(|| mf_frame.textures.nv12.texture.clone()), - _shared_handle: has_valid_zero_copy_handles.then_some(mf_frame.textures.nv12.handle), - _y_handle: has_valid_zero_copy_handles.then_some(mf_frame.textures.y.handle), - _uv_handle: has_valid_zero_copy_handles.then_some(mf_frame.textures.uv.handle), - nv12_data, - width: mf_frame.width, - height: mf_frame.height, - }; - - if !has_valid_zero_copy_handles { - decoder.recycle_textures(mf_frame.textures); - } - cached -} - enum Step { Seek(u32), Decode, diff --git a/crates/rendering/src/decoder/mod.rs b/crates/rendering/src/decoder/mod.rs index 1b378137f5f..419dc595644 100644 --- a/crates/rendering/src/decoder/mod.rs +++ b/crates/rendering/src/decoder/mod.rs @@ -115,20 +115,6 @@ impl SendableD3D11Texture { } } - pub fn new_with_yuv_handles( - texture: ID3D11Texture2D, - shared_handle: Option, - y_handle: Option, - uv_handle: Option, - ) -> Self { - Self { - texture, - shared_handle, - y_handle, - uv_handle, - } - } - pub fn inner(&self) -> &ID3D11Texture2D { &self.texture } @@ -490,34 +476,6 @@ impl DecodedFrame { } } - #[cfg(target_os = "windows")] - pub fn new_nv12_with_d3d11_texture_and_yuv_handles( - width: u32, - height: u32, - texture: ID3D11Texture2D, - shared_handle: Option, - y_handle: Option, - uv_handle: Option, - ) -> Self { - Self { - data: Arc::new(Vec::new()), - width, - height, - format: PixelFormat::Nv12, - matrix: YuvMatrix::Bt601, - y_stride: width, - uv_stride: width, - #[cfg(target_os = "macos")] - image_buf_backing: None, - d3d11_texture_backing: Some(Arc::new(SendableD3D11Texture::new_with_yuv_handles( - texture, - shared_handle, - y_handle, - uv_handle, - ))), - } - } - #[cfg(target_os = "windows")] #[allow(clippy::redundant_closure)] pub fn d3d11_texture_backing(&self) -> Option<&ID3D11Texture2D> { diff --git a/crates/video-decode/src/lib.rs b/crates/video-decode/src/lib.rs index 110d4b68730..9fb59896186 100644 --- a/crates/video-decode/src/lib.rs +++ b/crates/video-decode/src/lib.rs @@ -12,5 +12,5 @@ pub use ffmpeg::FFmpegDecoder; #[cfg(target_os = "windows")] pub use media_foundation::{ FrameTextures, MFDecodedFrame, MFDecoderCapabilities, MediaFoundationDecoder, NV12Data, - get_mf_decoder_capabilities, + NV12Sample, get_mf_decoder_capabilities, }; diff --git a/crates/video-decode/src/media_foundation.rs b/crates/video-decode/src/media_foundation.rs index 6b5755214bb..5b4de1c8067 100644 --- a/crates/video-decode/src/media_foundation.rs +++ b/crates/video-decode/src/media_foundation.rs @@ -1,3 +1,4 @@ +use std::collections::VecDeque; use std::path::Path; use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant}; @@ -143,6 +144,22 @@ pub struct MFDecodedFrame { pub plane_time: Duration, } +// The decoder's output surface for a sample: a texture array, the sample's slice in it, and its +// time. Media Foundation decodes into the slice again as soon as the sample is released, so the +// sample is held until the copy out of it is queued. +struct DecoderSurface { + _sample: IMFSample, + texture: ID3D11Texture2D, + subresource: u32, + pts: i64, +} + +pub struct NV12Sample { + pub pts: i64, + pub data: NV12Data, +} + +#[derive(Clone)] pub struct NV12Data { pub data: Arc>, pub y_stride: u32, @@ -514,6 +531,36 @@ fn create_internal_texture( }) } +unsafe fn create_staging_texture( + device: &ID3D11Device, + width: u32, + height: u32, +) -> Result { + 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 as u32, + MiscFlags: 0, + }; + + unsafe { + let mut tex: Option = None; + device + .CreateTexture2D(&desc, None, Some(&mut tex)) + .map_err(|e| format!("CreateTexture2D staging failed: {e:?}"))?; + tex.ok_or_else(|| "CreateTexture2D staging returned null".to_string()) + } +} + fn create_uav( device: &ID3D11Device, texture: &ID3D11Texture2D, @@ -536,6 +583,9 @@ 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; + pub struct MediaFoundationDecoder { source_reader: IMFSourceReader, d3d11_device: ID3D11Device, @@ -545,9 +595,10 @@ pub struct MediaFoundationDecoder { height: u32, frame_rate_num: u32, frame_rate_den: u32, - staging_texture: Option, - staging_width: u32, - staging_height: u32, + readback_free: Vec, + readback_queue: VecDeque<(i64, ID3D11Texture2D)>, + // How the stream stopped, reported once the queued readbacks are drained. + readback_end: Option>, frame_pool: FramePool, plane_converter: Nv12PlaneConverter, capabilities: MFDecoderCapabilities, @@ -606,6 +657,10 @@ impl MediaFoundationDecoder { 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::, _>>()?; + if width > capabilities.max_width || height > capabilities.max_height { warn!( video_width = width, @@ -633,9 +688,9 @@ impl MediaFoundationDecoder { height, frame_rate_num, frame_rate_den, - staging_texture: None, - staging_width: 0, - staging_height: 0, + readback_free, + readback_queue: VecDeque::new(), + readback_end: None, frame_pool: FramePool::new(), plane_converter, capabilities, @@ -663,60 +718,48 @@ impl MediaFoundationDecoder { &self.capabilities } - pub fn read_texture_to_cpu( - &mut self, - texture: &ID3D11Texture2D, - width: u32, - height: u32, - ) -> Result { - unsafe { self.read_texture_to_cpu_inner(texture, width, height) } - } - - unsafe fn read_texture_to_cpu_inner( - &mut self, - texture: &ID3D11Texture2D, - width: u32, - height: u32, - ) -> Result { - if self.staging_width != width - || self.staging_height != height - || self.staging_texture.is_none() + /// 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. + pub fn read_nv12(&mut self) -> Result, String> { + while self.readback_end.is_none() + && let Some(staging) = self.readback_free.pop() { - 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 as u32, - MiscFlags: 0, - }; - - let staging_texture = unsafe { - let mut tex: Option = None; - self.d3d11_device - .CreateTexture2D(&desc, None, Some(&mut tex)) - .map_err(|e| format!("CreateTexture2D staging failed: {e:?}"))?; - tex.ok_or("CreateTexture2D staging returned null")? - }; - - self.staging_texture = Some(staging_texture); - self.staging_width = width; - self.staging_height = height; + match unsafe { self.read_decoder_surface() } { + Ok(Some(surface)) => { + unsafe { + self.d3d11_context.CopySubresourceRegion( + &staging, + 0, + 0, + 0, + 0, + &surface.texture, + surface.subresource, + None, + ); + } + self.readback_queue.push_back((surface.pts, staging)); + } + end => { + self.readback_free.push(staging); + self.readback_end = Some(end.map(|_| ())); + } + } } - let staging = self.staging_texture.as_ref().unwrap(); - - unsafe { - self.d3d11_context.CopyResource(staging, texture); - } + let Some((pts, staging)) = self.readback_queue.pop_front() else { + return self + .readback_end + .take() + .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); + Ok(Some(NV12Sample { pts, data: data? })) + } + unsafe fn map_staging(&self, staging: &ID3D11Texture2D) -> Result { let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); unsafe { self.d3d11_context @@ -725,8 +768,8 @@ impl MediaFoundationDecoder { } let y_stride = mapped.RowPitch; - let y_height = height; - let uv_height = height / 2; + 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; @@ -757,7 +800,7 @@ impl MediaFoundationDecoder { } } - unsafe fn read_sample_inner(&mut self) -> Result, String> { + unsafe fn read_decoder_surface(&mut self) -> Result, String> { let mut stream_index = 0u32; let mut flags = 0u32; let mut timestamp = 0i64; @@ -819,6 +862,19 @@ impl MediaFoundationDecoder { .map_err(|e| format!("GetSubresourceIndex failed: {e:?}"))? }; + Ok(Some(DecoderSurface { + _sample: sample, + texture, + subresource: subresource_index, + pts: timestamp, + })) + } + + unsafe fn read_sample_inner(&mut self) -> Result, String> { + let Some(surface) = (unsafe { self.read_decoder_surface()? }) else { + return Ok(None); + }; + let frame_textures = self.frame_pool .acquire(&self.d3d11_device, self.width, self.height)?; @@ -832,8 +888,8 @@ impl MediaFoundationDecoder { 0, 0, 0, - &texture, - subresource_index, + &surface.texture, + surface.subresource, None, ); @@ -853,7 +909,7 @@ impl MediaFoundationDecoder { textures: Arc::new(frame_textures), width: self.width, height: self.height, - pts: timestamp, + pts: surface.pts, plane_time, })) } @@ -876,6 +932,9 @@ impl MediaFoundationDecoder { .map_err(|e| format!("Seek failed: {e:?}"))?; } + self.readback_free + .extend(self.readback_queue.drain(..).map(|(_, staging)| staging)); + self.readback_end = None; Ok(()) } } @@ -1136,6 +1195,59 @@ mod tests { } } + #[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"); + // 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 expected_end = loop { + match unsafe { decoder.read_decoder_surface() } { + Ok(Some(surface)) => unsafe { + decoder.d3d11_context.CopySubresourceRegion( + &staging, + 0, + 0, + 0, + 0, + &surface.texture, + surface.subresource, + None, + ); + expected.push((surface.pts, decoder.map_staging(&staging).unwrap().data)); + }, + end => break end.is_err(), + } + }; + drop(decoder); + + let mut decoder = MediaFoundationDecoder::new(&path).unwrap(); + let mut read = Vec::new(); + let end = loop { + match decoder.read_nv12() { + Ok(Some(sample)) => read.push((sample.pts, sample.data.data)), + end => break end.is_err(), + } + }; + assert!(expected.len() > READBACK_DEPTH); + assert!(read == expected); + assert_eq!(end, expected_end); + + let first = |decoder: &mut MediaFoundationDecoder| { + let sample = decoder.read_nv12().unwrap().unwrap(); + (sample.pts, sample.data.data) + }; + decoder.seek(0).unwrap(); + assert!(first(&mut decoder) == expected[0]); + decoder.read_nv12().unwrap(); + decoder.seek(0).unwrap(); + assert!(first(&mut decoder) == expected[0]); + } + fn normalized(path: &str) -> String { let encoded = media_foundation_path(Path::new(path)); assert_eq!(encoded.last(), Some(&0)); diff --git a/scripts/capbench.py b/scripts/capbench.py index 55fd98085be..47c94a642e0 100644 --- a/scripts/capbench.py +++ b/scripts/capbench.py @@ -80,6 +80,17 @@ def assign(self, process): ): raise OSError(ctypes.get_last_error(), "AssignProcessToJobObject failed") + def resume(self, process): + if self.handle is None: + return + from ctypes import wintypes + + ntdll = ctypes.WinDLL("ntdll") + ntdll.NtResumeProcess.argtypes = [wintypes.HANDLE] + status = ntdll.NtResumeProcess(wintypes.HANDLE(int(process._handle))) + if status != 0: + raise OSError(status, "NtResumeProcess failed") + def stats(self): if self.handle is None: return {} @@ -149,9 +160,13 @@ class JobAccounting(ctypes.Structure): ] +CREATE_SUSPENDED = 0x4 + + def run_measured(argv, timeout, env=None, cwd=None, affinity=None): job = Job(affinity) started = time.perf_counter() + # Started suspended so nothing it spawns can run before it is in the job. process = subprocess.Popen( argv, stdout=subprocess.PIPE, @@ -161,9 +176,16 @@ def run_measured(argv, timeout, env=None, cwd=None, affinity=None): text=True, encoding="utf-8", errors="replace", + creationflags=CREATE_SUSPENDED if job.handle is not None else 0, ) try: - job.assign(process) + try: + job.assign(process) + job.resume(process) + except BaseException: + process.kill() + process.wait() + raise lines = [] def pump(): @@ -451,6 +473,8 @@ def post(ctx, out_dir, metrics): "export-1080p60": export_bench(60, "1920x1080", "maximum"), "export-4k60": export_bench(60, "3840x2160", "maximum"), "export-720p30": export_bench(30, "1280x720", "social"), + # The editor decodes with FFmpeg on Windows (editor_instance.rs), so playback measures that path; + # Media Foundation is covered by decode-display and the exports. "playback-pipeline": example_bench( "playback-pipeline-benchmark", lambda ctx: ["--recording-path", str(ctx.fixture), "--fps", "60", "--frames", "300", "--force-ffmpeg"],