Merge branch 'master' into @wkazmierczak/WHIP-client

compositor_api/src/types/from_register_input.rs

@@ -81,6 +81,10 @@ impl TryFrom<InputRtpAudioOptions> for InputAudioStream {
     }
 }
 
+#[cfg(not(feature = "vk-video"))]
+const NO_VULKAN_VIDEO: &str =
+    "Requested `vulkan_video` decoder, but this binary was compiled without the `vk-video` feature.";
+
 impl TryFrom<RtpInput> for pipeline::RegisterInputOptions {
     type Error = TypeError;
 
@@ -106,10 +110,20 @@ impl TryFrom<RtpInput> for pipeline::RegisterInputOptions {
                 .as_ref()
                 .map(|video| {
                     Ok(input::rtp::InputVideoStream {
-                        options: match video {
-                            InputRtpVideoOptions::FfmepgH264 => decoder::VideoDecoderOptions {
+                        options: match video.decoder {
+                            VideoDecoder::FfmpegH264 => decoder::VideoDecoderOptions {
                                 decoder: pipeline::VideoDecoder::FFmpegH264,
                             },
+
+                            #[cfg(feature = "vk-video")]
+                            VideoDecoder::VulkanVideo => decoder::VideoDecoderOptions {
+                                decoder: pipeline::VideoDecoder::VulkanVideoH264,
+                            },
+
+                            #[cfg(not(feature = "vk-video"))]
+                            VideoDecoder::VulkanVideo => {
+                                return Err(TypeError::new(NO_VULKAN_VIDEO))
+                            }
                         },
                     })
                 })
@@ -146,6 +160,7 @@ impl TryFrom<Mp4Input> for pipeline::RegisterInputOptions {
             required,
             offset_ms,
             should_loop,
+            video_decoder,
         } = value;
 
         const BAD_URL_PATH_SPEC: &str =
@@ -165,10 +180,21 @@ impl TryFrom<Mp4Input> for pipeline::RegisterInputOptions {
             buffer_duration: None,
         };
 
+        let video_decoder = match video_decoder.unwrap_or(VideoDecoder::FfmpegH264) {
+            VideoDecoder::FfmpegH264 => pipeline::VideoDecoder::FFmpegH264,
+
+            #[cfg(feature = "vk-video")]
+            VideoDecoder::VulkanVideo => pipeline::VideoDecoder::VulkanVideoH264,
+
+            #[cfg(not(feature = "vk-video"))]
+            VideoDecoder::VulkanVideo => return Err(TypeError::new(NO_VULKAN_VIDEO)),
+        };
+
         Ok(pipeline::RegisterInputOptions {
             input_options: input::InputOptions::Mp4(input::mp4::Mp4Options {
                 source,
                 should_loop: should_loop.unwrap_or(false),
+                video_decoder,
             }),
             queue_options,
         })

compositor_api/src/types/register_input.rs

@@ -45,6 +45,8 @@ pub struct Mp4Input {
     /// Offset in milliseconds relative to the pipeline start (start request). If offset is
     /// not defined then stream is synchronized based on the first frames delivery time.
     pub offset_ms: Option<f64>,
+    /// (**default=`ffmpeg_h264`**) The decoder to use for decoding video.
+    pub video_decoder: Option<VideoDecoder>,
 }
 
 /// Capture streams from devices connected to Blackmagic DeckLink card.
@@ -122,8 +124,24 @@ pub enum InputRtpAudioOptions {
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
-#[serde(tag = "decoder", rename_all = "snake_case", deny_unknown_fields)]
-pub enum InputRtpVideoOptions {
-    #[serde(rename = "ffmpeg_h264")]
-    FfmepgH264,
+#[serde(rename_all = "snake_case", deny_unknown_fields)]
+pub struct InputRtpVideoOptions {
+    pub decoder: VideoDecoder,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, JsonSchema)]
+#[serde(rename_all = "snake_case", deny_unknown_fields)]
+pub enum VideoDecoder {
+    /// Use the software decoder based on ffmpeg.
+    FfmpegH264,
+
+    /// Use hardware decoder based on Vulkan Video.
+    ///
+    /// This should be faster and more scalable than teh ffmpeg decoder, if the hardware and OS
+    /// support it.
+    ///
+    /// This requires hardware that supports Vulkan Video. Another requirement is this program has
+    /// to be compiled with the `vk-video` feature enabled (enabled by default on platforms which
+    /// support Vulkan, i.e. non-Apple operating systems and not the web).
+    VulkanVideo,
 }

compositor_pipeline/Cargo.toml

@@ -29,6 +29,7 @@ tracing = { workspace = true }
 fdk-aac-sys = "0.5.0"
 rubato = "0.15.0"
 wgpu = { workspace = true }
+vk-video = { path = "../vk-video/", optional = true }
 glyphon = { workspace = true }
 webrtc = "0.11.0"
 tokio = {workspace = true }
@@ -38,7 +39,3 @@ url = "2.5.2"
 
 [target.x86_64-unknown-linux-gnu.dependencies]
 decklink = { path = "../decklink", optional = true }
-
-# platforms that support vulkan are: windows and all non-apple unixes. emscripten is a web-based platform, where vulkan is not available either
-[target.'cfg(any(windows, all(unix, not(target_os = "emscripten"), not(target_os = "ios"), not(target_os = "macos"))))'.dependencies]
-vk-video = { path = "../vk-video/", optional = true }

compositor_pipeline/src/pipeline/input/mp4.rs

@@ -6,7 +6,10 @@ use crossbeam_channel::Receiver;
 use tracing::error;
 
 use crate::{
-    pipeline::decoder::{AudioDecoderOptions, VideoDecoderOptions},
+    pipeline::{
+        decoder::{AudioDecoderOptions, VideoDecoderOptions},
+        VideoDecoder,
+    },
     queue::PipelineEvent,
 };
 
@@ -20,6 +23,7 @@ pub mod mp4_file_reader;
 pub struct Mp4Options {
     pub source: Source,
     pub should_loop: bool,
+    pub video_decoder: VideoDecoder,
 }
 
 pub(crate) enum Mp4ReaderOptions {
@@ -96,13 +100,16 @@ impl Mp4 {
                 should_loop: options.should_loop,
             },
             input_id.clone(),
+            options.video_decoder,
         )?;
 
         let (video_reader, video_receiver) = match video {
             Some((reader, receiver)) => {
                 let input_receiver = VideoInputReceiver::Encoded {
                     chunk_receiver: receiver,
-                    decoder_options: reader.decoder_options(),
+                    decoder_options: VideoDecoderOptions {
+                        decoder: options.video_decoder,
+                    },
                 };
                 (Some(reader), Some(input_receiver))
             }

compositor_pipeline/src/pipeline/input/mp4/mp4_file_reader.rs

@@ -125,6 +125,7 @@ impl Mp4FileReader<VideoDecoderOptions> {
     pub(crate) fn new_video(
         options: Mp4ReaderOptions,
         input_id: InputId,
+        video_decoder: VideoDecoder,
     ) -> Result<Option<(Mp4FileReader<VideoDecoderOptions>, ChunkReceiver)>, Mp4Error> {
         let stop_thread = Arc::new(AtomicBool::new(false));
         let span = span!(Level::INFO, "MP4 video", input_id = input_id.to_string());
@@ -136,7 +137,7 @@ impl Mp4FileReader<VideoDecoderOptions> {
                 Self::new(
                     input_file,
                     size,
-                    Self::find_h264_info,
+                    |r| Self::find_h264_info(r, video_decoder),
                     None,
                     stop_thread,
                     span,
@@ -153,7 +154,7 @@ impl Mp4FileReader<VideoDecoderOptions> {
                 Self::new(
                     reader,
                     size,
-                    Self::find_h264_info,
+                    |r| Self::find_h264_info(r, video_decoder),
                     Some(fragment_receiver),
                     stop_thread,
                     span,
@@ -165,6 +166,7 @@ impl Mp4FileReader<VideoDecoderOptions> {
 
     fn find_h264_info<Reader: Read + Seek + Send + 'static>(
         reader: &mp4::Mp4Reader<Reader>,
+        video_decoder: VideoDecoder,
     ) -> Option<TrackInfo<VideoDecoderOptions, impl FnMut(mp4::Mp4Sample) -> Bytes>> {
         let (&track_id, track, avc) = reader.tracks().iter().find_map(|(id, track)| {
             let track_type = track.track_type().ok()?;
@@ -234,7 +236,7 @@ impl Mp4FileReader<VideoDecoderOptions> {
         };
 
         let decoder_options = VideoDecoderOptions {
-            decoder: VideoDecoder::FFmpegH264,
+            decoder: video_decoder,
         };
 
         Some(TrackInfo {

compositor_pipeline/src/pipeline/output/whip/init_peer_connection.rs

@@ -8,7 +8,7 @@ use std::{
     env::{self, VarError},
     sync::Arc,
 };
-use tracing::{error, info, warn};
+use tracing::{info, warn};
 use webrtc::{
     api::{
         interceptor_registry::register_default_interceptors,

integration_tests/examples/encoded_channel_output.rs

@@ -91,6 +91,7 @@ fn main() {
         input_options: InputOptions::Mp4(Mp4Options {
             source: Source::File(root_dir.join(BUNNY_FILE_PATH)),
             should_loop: false,
+            video_decoder: compositor_pipeline::pipeline::VideoDecoder::FFmpegH264,
         }),
         queue_options: QueueInputOptions {
             required: true,

integration_tests/examples/mp4_hw_dec.rs

@@ -0,0 +1,66 @@
+use anyhow::Result;
+use compositor_api::types::Resolution;
+use serde_json::json;
+use std::time::Duration;
+
+use integration_tests::{
+    examples::{self, run_example},
+    ffmpeg::start_ffmpeg_receive,
+};
+
+const VIDEO_RESOLUTION: Resolution = Resolution {
+    width: 1280,
+    height: 720,
+};
+
+const IP: &str = "127.0.0.1";
+const OUTPUT_PORT: u16 = 8004;
+
+fn main() {
+    run_example(client_code);
+}
+
+fn client_code() -> Result<()> {
+    start_ffmpeg_receive(Some(OUTPUT_PORT), None)?;
+
+    examples::post(
+        "input/input_1/register",
+        &json!({
+            "type": "mp4",
+            "path": "examples/assets/BigBuckBunny.mp4",
+            "video_decoder": "vulkan_video",
+        }),
+    )?;
+
+    examples::post(
+        "output/output_1/register",
+        &json!({
+            "type": "rtp_stream",
+            "port": OUTPUT_PORT,
+            "ip": IP,
+            "video": {
+                "resolution": {
+                    "width": VIDEO_RESOLUTION.width,
+                    "height": VIDEO_RESOLUTION.height,
+                },
+                "encoder": {
+                    "type": "ffmpeg_h264",
+                    "preset": "ultrafast"
+                },
+                "initial": {
+                    "root": {
+                        "id": "input_1",
+                        "type": "input_stream",
+                        "input_id": "input_1",
+                    }
+                }
+            }
+        }),
+    )?;
+
+    std::thread::sleep(Duration::from_millis(500));
+
+    examples::post("start", &json!({}))?;
+
+    Ok(())
+}

integration_tests/examples/raw_channel_output.rs

@@ -113,6 +113,7 @@ fn main() {
         input_options: InputOptions::Mp4(Mp4Options {
             source: Source::File(root_dir().join(BUNNY_FILE_PATH)),
             should_loop: false,
+            video_decoder: compositor_pipeline::pipeline::VideoDecoder::FFmpegH264,
         }),
         queue_options: QueueInputOptions {
             required: true,