diff --git a/base/cvd/cuttlefish/host/commands/run_cvd/launch/vhost_user_media_devices.cpp b/base/cvd/cuttlefish/host/commands/run_cvd/launch/vhost_user_media_devices.cpp index 420b10f5a02..71b877d002b 100644 --- a/base/cvd/cuttlefish/host/commands/run_cvd/launch/vhost_user_media_devices.cpp +++ b/base/cvd/cuttlefish/host/commands/run_cvd/launch/vhost_user_media_devices.cpp @@ -17,6 +17,7 @@ #include +#include #include #include #include @@ -67,23 +68,44 @@ class VhostUserMediaDevices : public CommandSource { for (int index = 0; index < instance_.media_configs().size(); index++) { auto config = instance_.media_configs()[index]; std::string binary_path; + std::optional cmd; if (config.type == CuttlefishConfig::MediaType::kV4l2EmulatedCameraMPlane) { binary_path = VhostUserMediaEmulatedCameraMPlaneBinary(); + cmd.emplace(NewCommand(binary_path, instance_.media_socket_path(index), + config.lens_facing)); } else if (config.type == CuttlefishConfig::MediaType::kV4l2EmulatedCameraSPlane) { binary_path = VhostUserMediaEmulatedCameraSPlaneBinary(); + cmd.emplace(NewCommand(binary_path, instance_.media_socket_path(index), + config.lens_facing)); + } else if (config.type == CuttlefishConfig::MediaType::kV4l2StreamProxy) { + CF_EXPECT(config.v4l2_stream_proxy.has_value(), + "Missing v4l2_stream_proxy config"); + binary_path = VhostUserMediaV4l2StreamProxyBinary(); + cmd.emplace( + NewCommand(binary_path, instance_.media_socket_path(index), "")); + cmd->AddParameter("--input_path=", + config.v4l2_stream_proxy->input_path); + cmd->AddParameter( + "--input_width=", + std::to_string(config.v4l2_stream_proxy->input_width)); + cmd->AddParameter( + "--input_height=", + std::to_string(config.v4l2_stream_proxy->input_height)); + cmd->AddParameter("--input_fps=", config.v4l2_stream_proxy->input_fps); } else if (config.type == CuttlefishConfig::MediaType::kV4l2Proxy) { continue; } else { CF_EXPECT(false, "unknown media type"); } - Command cmd = NewCommand(binary_path, instance_.media_socket_path(index), - config.lens_facing); + + CF_EXPECT(cmd.has_value(), "Command was not initialized"); + Command cmd_log_tee = CF_EXPECT( - log_tee_.CreateLogTee(cmd, "vhu_media_simple_device", kStdErr), + log_tee_.CreateLogTee(*cmd, "vhu_media_simple_device", kStdErr), "Failed to create log tee command for media device"); - commands.emplace_back(std::move(cmd)); + commands.emplace_back(std::move(*cmd)); commands.emplace_back(std::move(cmd_log_tee)); } return commands; diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/Cargo.toml b/base/cvd/cuttlefish/host/commands/vhost_user_media/Cargo.toml index 6753d3ba291..2f5b2e3e7ad 100644 --- a/base/cvd/cuttlefish/host/commands/vhost_user_media/Cargo.toml +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/Cargo.toml @@ -3,6 +3,7 @@ members = [ "emulated_camera_mplane", "emulated_camera_splane", "vhu_media", + "v4l2_stream_proxy", ] [workspace.dependencies] @@ -11,6 +12,7 @@ env_logger = "0.11" image = "0.24" libc = "0.2" log = "0.4" +nix = "0.29.0" thiserror = "2.0" v4l2r = { version = "0.0.6", features = ["arch64"] } vhost = { version = "0.16.0", features = ["vhost-user-backend"] } diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/BUILD.bazel b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/BUILD.bazel new file mode 100644 index 00000000000..5574082df74 --- /dev/null +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/BUILD.bazel @@ -0,0 +1,47 @@ +load("@rules_rust//rust:defs.bzl", "rust_binary") +load("@vhost_user_media_workspace_crates//:defs.bzl", "crate_deps") + +package(default_visibility = ["//:android_cuttlefish"]) + +alias( + name = "v4l2_stream_proxy", + actual = select({ + "@platforms//cpu:x86_64": ":v4l2_stream_proxy_binary", + "//conditions:default": ":unsupported_binary", + }), +) + +rust_binary( + name = "v4l2_stream_proxy_binary", + srcs = [ + "src/device.rs", + "src/main.rs", + "src/worker.rs", + ], + edition = "2024", + deps = crate_deps([ + "clap", + "env_logger", + "libc", + "log", + "nix", + "v4l2r", + "vhost-user-backend", + "virtio-media", + "virtio-queue", + "vm-memory", + ]) + [ + "//cuttlefish/host/commands/vhost_user_media/vhu_media", + ], +) + +genrule( + name = "unsupported_binary", + outs = ["unsupported.sh"], + cmd = """cat << 'EOF' > $@ +#!/bin/bash +echo "platform unsupported" >&2 +exit 1 +EOF""", + executable = True, +) diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/Cargo.toml b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/Cargo.toml new file mode 100644 index 00000000000..1558d6bae55 --- /dev/null +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "v4l2_stream_proxy" +version = "0.1.0" +edition = "2024" + +[dependencies] +clap = { workspace = true } +env_logger = { workspace = true } +libc = { workspace = true } +log = { workspace = true } +nix = { workspace = true, features = ["event", "poll"] } +v4l2r = { workspace = true } +vhost-user-backend = { workspace = true } +vhu_media = { workspace = true } +virtio-media = { workspace = true } +virtio-queue = { workspace = true } +vm-memory = { workspace = true } diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/README.md b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/README.md new file mode 100644 index 00000000000..5cc1ae2ba77 --- /dev/null +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/README.md @@ -0,0 +1,70 @@ +# V4L2 Stream Proxy Host Tool + +This directory contains the `v4l2_stream_proxy` vhost-user device. To test this device, you can use the provided helper script to start a video stream on the host, which can then be proxied by the `cvd` tool from the host to the guest. + +The test helper script is located at: +`tools/testutils/ffmpeg_v4l2_stream_proxy.sh` (relative to the repository root). + +## Prerequisites + +The script requires `ffmpeg` (and `ffprobe` if streaming from a file) to be installed on the host system. + +```bash +sudo apt install ffmpeg +``` + +## Usage + +The script creates a FIFO (named pipe) and runs `ffmpeg` in a loop to continuously feed video into it. When `ffmpeg` exits (or fails), the script automatically restarts it. `ffmpeg`'s normal behavior includes exiting when the FIFO's reader closes the FIFO. This happens whenever the camera's stream is stopped in the guest. + +In both modes (Test Source and Video File), the script will print an argument to add to `cvd create`/`cvd start` that will provide a video device in the guest that matches the host configuration. + +### 1. Test Source Mode (Default) + +If no video file is provided, the script uses `ffmpeg`'s `testsrc` filter to generate a synthetic test pattern (color bars and a timer). + +To start streaming a 640x480 video at 30 fps (defaults): + +```bash +tools/testutils/ffmpeg_v4l2_stream_proxy.sh +``` + +You can customize the resolution and frame rate: + +```bash +tools/testutils/ffmpeg_v4l2_stream_proxy.sh -w 1280 -H 720 -r 60 +``` + +Available options: +* `-w `: Video width (default: 640) +* `-H `: Video height (default: 480) +* `-r `: Video FPS (default: 30) +* `-p `: Path to the FIFO to create/use (default: `/tmp/v4l2_fifo`) + +### 2. Video File Mode + +To stream a video file (e.g., `.mp4`, `.mjpeg`, `.h264`) in a loop: + +```bash +tools/testutils/ffmpeg_v4l2_stream_proxy.sh -f /path/to/video.mp4 +``` + +In this mode, the script uses `ffprobe` to automatically detect the video's resolution and frame rate. + +Available options: +* `-f `: Path to the host video file +* `-p `: Path to the FIFO to create/use (default: `/tmp/v4l2_fifo`) + +## Connecting to Cuttlefish (CVD) + +Once the streaming script is running, start Cuttlefish and configure it to use the stream. Use the parameters printed by the script (or the defaults) to configure the `--media` flag. + +Example CVD launch command: + +```bash +cvd create --media=v4l2_stream_proxy:input_path=/tmp/v4l2_fifo:input_width=640:input_height=480:input_fps=30 +``` + +## Stopping the Stream + +Press `Ctrl+C` in the terminal running the script. The script will catch the interrupt, kill the active `ffmpeg` process, and clean up the FIFO if it was created by the script. diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/device.rs b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/device.rs new file mode 100644 index 00000000000..e7a72332b80 --- /dev/null +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/device.rs @@ -0,0 +1,802 @@ +// Copyright 2026, The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::VecDeque; +use std::io::Result as IoResult; +use std::os::fd::AsFd; +use std::os::fd::BorrowedFd; +use std::sync::{Arc, Mutex}; +use std::sync::mpsc::channel; +use std::time::Instant; +use nix::sys::eventfd::{EventFd, EfdFlags}; + + +use v4l2r::PixelFormat; +use v4l2r::QueueType; +use v4l2r::bindings; +use v4l2r::bindings::v4l2_fmtdesc; +use v4l2r::bindings::v4l2_format; +use v4l2r::bindings::v4l2_requestbuffers; +use v4l2r::ioctl::BufferCapabilities; +use v4l2r::ioctl::BufferField; +use v4l2r::ioctl::BufferFlags; +use v4l2r::ioctl::V4l2Buffer; +use v4l2r::ioctl::V4l2PlanesWithBackingMut; +use v4l2r::memory::MemoryType; +use virtio_media::VirtioMediaDevice; +use virtio_media::VirtioMediaDeviceSession; +use virtio_media::VirtioMediaEventQueue; +use virtio_media::VirtioMediaHostMemoryMapper; +use virtio_media::io::ReadFromDescriptorChain; +use virtio_media::io::WriteToDescriptorChain; +use virtio_media::ioctl::IoctlResult; +use virtio_media::ioctl::VirtioMediaIoctlHandler; +use virtio_media::ioctl::virtio_media_dispatch_ioctl; +use virtio_media::memfd::MemFdBuffer; +use virtio_media::mmap::MmapMappingManager; +use virtio_media::protocol::SgEntry; +use virtio_media::protocol::V4l2Ioctl; +use virtio_media::protocol::VIRTIO_MEDIA_MMAP_FLAG_RW; + +use crate::Config; +use crate::worker::{WorkerCmd, WorkerHandle, worker_thread_loop}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Format { + Yuv420M, +} + +impl Format { + pub fn from_str(s: &str) -> Option { + match s { + "YUV420M" => Some(Self::Yuv420M), + _ => None, + } + } + + pub fn as_str(&self) -> &'static str { + match self { + Self::Yuv420M => "YUV420M", + } + } + + pub fn fourcc(&self) -> u32 { + let fourcc_bytes = match self { + Self::Yuv420M => b"YM12", + }; + PixelFormat::from_fourcc(fourcc_bytes).to_u32() + } + + pub fn bytesperline(&self, width: u32, plane_idx: usize) -> u32 { + match self { + Self::Yuv420M => { + if plane_idx == 0 { + width + } else { + width / 2 + } + } + } + } + + pub fn plane_sizes(&self, width: u32, height: u32) -> Vec { + let w = width as usize; + let h = height as usize; + match self { + Self::Yuv420M => vec![ + w * h, + w * h / 4, + w * h / 4, + ], + } + } + + pub fn frame_size(&self, width: u32, height: u32) -> usize { + self.plane_sizes(width, height).iter().sum() + } +} + +/// Current status of a buffer. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum BufferState { + /// Buffer has just been created (or streamed off) and not been used yet. + New, + /// Buffer has been QBUF'd by the driver but not yet processed. + Incoming, + /// Buffer has been processed and is ready for dequeue. + Outgoing { + /// Sequence of the generated frame. + sequence: u32, + }, +} + +/// Information about a single plane of a multi-planar buffer. +pub(crate) struct Plane { + /// Backing memory file descriptor. + pub(crate) fd: MemFdBuffer, + /// Offset that can be used to map the plane's memory. + offset: u32, +} + +/// Information about a single buffer. +pub(crate) struct Buffer { + /// Current state of the buffer. + pub(crate) state: BufferState, + /// V4L2 representation of this buffer to be sent to the guest when requested. + pub(crate) v4l2_buffer: V4l2Buffer, + /// Backing storage and offsets for the planes of the buffer. + pub(crate) planes: Vec, +} + +impl Buffer { + fn new(v4l2_buffer: V4l2Buffer, planes: Vec) -> Self { + Self { + state: BufferState::New, + v4l2_buffer, + planes, + } + } + + fn unset_flag(flags: &mut BufferFlags, v: BufferFlags) { + *flags &= !v; + } + + /// Update the state of the buffer as well as its V4L2 representation. + pub(crate) fn set_state(&mut self, state: BufferState) { + let mut flags = self.v4l2_buffer.flags(); + match state { + BufferState::New => { + self.clear_bytesused(); + Self::unset_flag(&mut flags, BufferFlags::QUEUED); + } + BufferState::Incoming => { + self.clear_bytesused(); + flags |= BufferFlags::QUEUED; + } + BufferState::Outgoing { sequence } => { + self.v4l2_buffer.set_sequence(sequence); + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: clock_gettime is a standard POSIX libc call with a valid pointer. + unsafe { + libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts); + } + self.v4l2_buffer.set_timestamp(bindings::timeval { + tv_sec: ts.tv_sec as bindings::__time_t, + tv_usec: (ts.tv_nsec / 1000) as bindings::__time_t, + }); + Self::unset_flag(&mut flags, BufferFlags::QUEUED); + + // Set bytesused to plane length for all planes + if let V4l2PlanesWithBackingMut::Mmap(planes) = self.v4l2_buffer.planes_with_backing_iter_mut() { + for mut plane in planes { + let len = *plane.length; + *plane.bytesused = len; + } + } + } + } + self.v4l2_buffer.set_flags(flags); + self.state = state; + } + + fn clear_bytesused(&mut self) { + if let V4l2PlanesWithBackingMut::Mmap(planes) = self.v4l2_buffer.planes_with_backing_iter_mut() { + for mut plane in planes { + *plane.bytesused = 0; + } + } + } +} + +/// Inner state of a session. +pub(crate) struct SessionState { + /// Id of the session. + pub(crate) id: u32, + /// Buffers currently allocated for this session. + pub(crate) buffers: Vec, + /// Queue of buffers awaiting processing. + pub(crate) queued_buffers: VecDeque, + /// Current sequence number of the generated frames. + pub(crate) sequence: u32, + /// Time when the last frame was completed. + pub(crate) last_frame_time: Instant, +} + +/// Session data of [`V4l2Stream`]. +pub struct V4l2StreamSession { + /// Id of the session. + id: u32, + state: Arc>, + worker: Option, + /// Is the session currently streaming? + streaming: bool, +} + +impl VirtioMediaDeviceSession for V4l2StreamSession { + fn poll_fd(&self) -> Option> { + None + } +} + +/// V4l2 stream device used for testing Android camera stack. +/// +/// This implementation looks forward to have feature parity with existing Android Guest Emulated +/// Camera HAL. +pub struct V4l2Stream { + /// Queue used to send events to the guest. + evt_queue: Arc>, + /// Host MMAP mapping manager. + mmap_manager: MmapMappingManager, + /// ID of the session with allocated buffers, if any. + /// + /// v4l2-compliance checks that only a single session can have allocated buffers at a given + /// time, since that's how actual hardware works - no two sessions can access a camera at the + /// same time. It will fails if we allow simultaneous sessions to be active, so we need this + /// artificial limitation to make it pass fully. + active_session: Option, + config: Config, +} + +impl V4l2Stream +where + Q: VirtioMediaEventQueue + Send + 'static, + HM: VirtioMediaHostMemoryMapper, +{ + pub fn new(evt_queue: Q, mapper: HM, config: Config) -> Self { + Self { + evt_queue: Arc::new(Mutex::new(evt_queue)), + mmap_manager: MmapMappingManager::from(mapper), + active_session: None, + config, + } + } + + fn queue_type(&self) -> QueueType { + QueueType::VideoCaptureMplane + } + + fn default_fmt(&self, queue: QueueType) -> v4l2_format { + let plane_sizes = self.config.format.plane_sizes(self.config.input_width, self.config.input_height); + let mut plane_fmt: [bindings::v4l2_plane_pix_format; 8] = Default::default(); + for (i, &size) in plane_sizes.iter().enumerate() { + plane_fmt[i].sizeimage = size as u32; + plane_fmt[i].bytesperline = self.config.format.bytesperline(self.config.input_width, i); + } + + let pix_mp = bindings::v4l2_pix_format_mplane { + width: self.config.input_width, + height: self.config.input_height, + pixelformat: self.config.format.fourcc(), + field: bindings::v4l2_field_V4L2_FIELD_NONE, + colorspace: bindings::v4l2_colorspace_V4L2_COLORSPACE_SRGB, + num_planes: plane_sizes.len() as u8, + plane_fmt, + ..Default::default() + }; + v4l2_format { + type_: queue as u32, + fmt: bindings::v4l2_format__bindgen_ty_1 { pix_mp }, + } + } + + fn default_fmtdesc(&self, queue: QueueType) -> v4l2_fmtdesc { + let mut fmtdesc: bindings::v4l2_fmtdesc = Default::default(); + fmtdesc.type_ = queue as u32; + fmtdesc.pixelformat = self.config.format.fourcc(); + let desc = self.config.format.as_str().as_bytes(); + fmtdesc.description[0..desc.len()].copy_from_slice(desc); + fmtdesc + } +} + +impl VirtioMediaDevice for V4l2Stream +where + Q: VirtioMediaEventQueue + Send + 'static, + HM: VirtioMediaHostMemoryMapper, + Reader: ReadFromDescriptorChain, + Writer: WriteToDescriptorChain, +{ + type Session = V4l2StreamSession; + + fn new_session(&mut self, session_id: u32) -> std::result::Result { + Ok(V4l2StreamSession { + id: session_id, + state: Arc::new(Mutex::new(SessionState { + id: session_id, + buffers: Vec::new(), + queued_buffers: VecDeque::new(), + sequence: 0, + last_frame_time: Instant::now(), + })), + worker: None, + streaming: false, + }) + } + + fn close_session(&mut self, mut session: Self::Session) { + if session.streaming { + let _ = self.streamoff(&mut session, self.queue_type()); + } + if let Some(id) = self.active_session { + if id == session.id { + self.active_session = None; + } + } + + let state = session.state.lock().unwrap(); + if state.buffers.is_empty() { + return; + } + + for buffer in &state.buffers { + for plane in &buffer.planes { + self.mmap_manager.unregister_buffer(plane.offset); + } + } + } + + fn do_ioctl( + &mut self, + session: &mut Self::Session, + ioctl: V4l2Ioctl, + reader: &mut Reader, + writer: &mut Writer, + ) -> IoResult<()> { + virtio_media_dispatch_ioctl(self, session, ioctl, reader, writer) + } + + fn do_mmap( + &mut self, + session: &mut Self::Session, + flags: u32, + offset: u32, + ) -> std::result::Result<(u64, u64), i32> { + let mut state = session.state.lock().unwrap(); + // Find which plane in which buffer matches the offset + let mut found_plane: Option<(&mut Plane, u32)> = None; + for buffer in &mut state.buffers { + if let Some(idx) = buffer.planes.iter().position(|p| p.offset == offset) { + found_plane = Some((&mut buffer.planes[idx], offset)); + break; + } + } + let (plane, offset) = found_plane.ok_or(libc::EINVAL)?; + let rw = (flags & VIRTIO_MEDIA_MMAP_FLAG_RW) != 0; + let fd = plane.fd.as_file().as_fd(); + let (guest_addr, size) = self + .mmap_manager + .create_mapping(offset, fd, rw) + .map_err(|_| libc::EINVAL)?; + Ok((guest_addr, size)) + } + + fn do_munmap(&mut self, guest_addr: u64) -> std::result::Result<(), i32> { + self.mmap_manager + .remove_mapping(guest_addr) + .map(|_| ()) + .map_err(|_| libc::EINVAL) + } +} + +const INPUTS: [bindings::v4l2_input; 1] = [bindings::v4l2_input { + index: 0, + name: *b"Default\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", + type_: bindings::V4L2_INPUT_TYPE_CAMERA, + ..unsafe { std::mem::zeroed() } +}]; + +impl VirtioMediaIoctlHandler for V4l2Stream +where + Q: VirtioMediaEventQueue + Send + 'static, + HM: VirtioMediaHostMemoryMapper, +{ + type Session = V4l2StreamSession; + + fn enum_fmt( + &mut self, + _session: &Self::Session, + queue: QueueType, + index: u32, + ) -> IoctlResult { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + if index > 0 { + return Err(libc::EINVAL); + } + Ok(self.default_fmtdesc(queue)) + } + + fn g_fmt( + &mut self, + _session: &Self::Session, + queue: QueueType, + ) -> IoctlResult { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + Ok(self.default_fmt(queue)) + } + + fn s_fmt( + &mut self, + session: &mut Self::Session, + queue: QueueType, + format: v4l2_format, + ) -> IoctlResult { + self.try_fmt(session, queue, format) + } + + fn try_fmt( + &mut self, + _session: &Self::Session, + queue: QueueType, + _format: v4l2_format, + ) -> IoctlResult { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + Ok(self.default_fmt(queue)) + } + + fn g_parm( + &mut self, + _session: &Self::Session, + queue: QueueType, + ) -> IoctlResult { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + + let mut parm = bindings::v4l2_streamparm { + type_: queue as u32, + ..Default::default() + }; + + let (numerator, denominator) = self.config.fps_interval; + + // SAFETY: The `parm` union is used for the capture type. + let capture = unsafe { &mut parm.parm.capture }; + capture.capability = bindings::V4L2_CAP_TIMEPERFRAME; + capture.timeperframe = bindings::v4l2_fract { + numerator, + denominator, + }; + + Ok(parm) + } + + fn s_parm( + &mut self, + _session: &mut Self::Session, + mut parm: bindings::v4l2_streamparm, + ) -> IoctlResult { + if parm.type_ != self.queue_type() as u32 { + return Err(libc::EINVAL); + } + + let (numerator, denominator) = self.config.fps_interval; + + // We just return the fixed values, ignoring what the user set. + // SAFETY: The `parm` union is used for the capture type. + let capture = unsafe { &mut parm.parm.capture }; + capture.capability = bindings::V4L2_CAP_TIMEPERFRAME; + capture.timeperframe = bindings::v4l2_fract { + numerator, + denominator, + }; + + Ok(parm) + } + + fn reqbufs( + &mut self, + session: &mut Self::Session, + queue: QueueType, + memory: MemoryType, + count: u32, + ) -> IoctlResult { + let expected_queue = self.queue_type(); + if queue != expected_queue { + return Err(libc::EINVAL); + } + if memory != MemoryType::Mmap { + return Err(libc::EINVAL); + } + if session.streaming { + return Err(libc::EBUSY); + } + match self.active_session { + Some(id) if id != session.id => return Err(libc::EBUSY), + _ => (), + } + + let mut state = session.state.lock().unwrap(); + + if count == 0 { + self.active_session = None; + state.queued_buffers.clear(); + for buffer in state.buffers.iter_mut() { + buffer.set_state(BufferState::New); + } + } else { + state.queued_buffers.clear(); + for buffer in state.buffers.iter_mut() { + buffer.set_state(BufferState::New); + } + self.active_session = Some(session.id); + } + + let count = std::cmp::min(count, 32); + + for buffer in &state.buffers { + for plane in &buffer.planes { + self.mmap_manager.unregister_buffer(plane.offset); + } + } + + let plane_sizes = self.config.format.plane_sizes(self.config.input_width, self.config.input_height); + let num_planes = plane_sizes.len(); + + state.buffers = (0..count) + .map(|i| -> Result { + let mut planes = Vec::new(); + + for &size in &plane_sizes { + let fd = MemFdBuffer::new(size as u64) + .map_err(|e| { + log::error!("failed to allocate MMAP buffer: {:#}", e); + libc::ENOMEM + })?; + let offset = self + .mmap_manager + .register_buffer(None, size as u32) + .map_err(|_| libc::EINVAL)?; + planes.push(Plane { fd, offset }); + } + + let mut v4l2_buffer = V4l2Buffer::new(expected_queue, i, MemoryType::Mmap); + + if num_planes > 1 { + unsafe { + (*v4l2_buffer.as_mut_ptr()).length = num_planes as u32; + } + if let V4l2PlanesWithBackingMut::Mmap(planes_iter) = + v4l2_buffer.planes_with_backing_iter_mut() + { + for (j, mut plane) in planes_iter.enumerate() { + plane.set_mem_offset(planes[j].offset); + *plane.length = plane_sizes[j] as u32; + } + } else { + panic!() + } + } else { + if let V4l2PlanesWithBackingMut::Mmap(mut planes_iter) = + v4l2_buffer.planes_with_backing_iter_mut() + { + let mut plane = planes_iter.next().unwrap(); + plane.set_mem_offset(planes[0].offset); + *plane.length = plane_sizes[0] as u32; + } else { + panic!() + } + } + + v4l2_buffer.set_field(BufferField::None); + v4l2_buffer.set_flags(BufferFlags::TIMESTAMP_MONOTONIC); + + Ok(Buffer::new(v4l2_buffer, planes)) + }) + .collect::>()?; + + Ok(v4l2_requestbuffers { + count, + type_: queue as u32, + memory: memory as u32, + capabilities: (BufferCapabilities::SUPPORTS_MMAP + | BufferCapabilities::SUPPORTS_ORPHANED_BUFS) + .bits(), + flags: 0, + ..Default::default() + }) + } + + fn querybuf( + &mut self, + session: &V4l2StreamSession, + queue: QueueType, + index: u32, + ) -> IoctlResult { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + let state = session.state.lock().unwrap(); + let buffer = state.buffers.get(index as usize).ok_or(libc::EINVAL)?; + Ok(buffer.v4l2_buffer.clone()) + } + + fn qbuf( + &mut self, + session: &mut Self::Session, + qbuf: V4l2Buffer, + _sg_entries: Vec>, + ) -> IoctlResult { + let mut state = session.state.lock().unwrap(); + let buf_id = qbuf.index() as usize; + + let buf_v4l2 = { + let buffer = state.buffers.get_mut(buf_id).ok_or(libc::EINVAL)?; + if buffer.state == BufferState::Incoming { + return Err(libc::EINVAL); + } + buffer.set_state(BufferState::Incoming); + buffer.v4l2_buffer.clone() + }; + + state.queued_buffers.push_back(buf_id); + + if session.streaming { + if let Some(ref worker) = session.worker { + let _ = worker.tx.send(WorkerCmd::BufferQueued); + let _ = worker.event_fd.write(1); + } + } + + Ok(buf_v4l2) + } + + fn streamon(&mut self, session: &mut Self::Session, queue: QueueType) -> IoctlResult<()> { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + + { + let mut state = session.state.lock().unwrap(); + + if state.buffers.is_empty() { + return Err(libc::EINVAL); + } + state.sequence = 0; + } + + if session.streaming { + return Ok(()); + } + session.streaming = true; + + let event_fd = Arc::new( + EventFd::from_flags(EfdFlags::EFD_CLOEXEC | EfdFlags::EFD_NONBLOCK) + .map_err(|_| libc::ENOMEM)?, + ); + let event_fd_clone = Arc::clone(&event_fd); + + let (tx, rx) = channel(); + let state_clone = Arc::clone(&session.state); + let evt_queue_clone = Arc::clone(&self.evt_queue); + let config_clone = self.config.clone(); + + let join_handle = std::thread::spawn(move || { + worker_thread_loop(config_clone, evt_queue_clone, state_clone, rx, event_fd_clone); + }); + + session.worker = Some(WorkerHandle { tx, event_fd, join_handle }); + + Ok(()) + } + + fn streamoff(&mut self, session: &mut Self::Session, queue: QueueType) -> IoctlResult<()> { + if queue != self.queue_type() { + return Err(libc::EINVAL); + } + if session.streaming { + session.streaming = false; + if let Some(worker) = session.worker.take() { + let _ = worker.tx.send(WorkerCmd::Stop); + let _ = worker.event_fd.write(1); + let _ = worker.join_handle.join(); + } + } + + let mut state = session.state.lock().unwrap(); + state.queued_buffers.clear(); + for buffer in state.buffers.iter_mut() { + buffer.set_state(BufferState::New); + } + + Ok(()) + } + + fn g_input(&mut self, _session: &Self::Session) -> IoctlResult { + Ok(0) + } + + fn s_input(&mut self, _session: &mut Self::Session, input: i32) -> IoctlResult { + if input != 0 { Err(libc::EINVAL) } else { Ok(0) } + } + + fn enuminput( + &mut self, + _session: &Self::Session, + index: u32, + ) -> IoctlResult { + INPUTS.get(index as usize).copied().ok_or(libc::EINVAL) + } + + fn enum_framesizes( + &mut self, + _session: &Self::Session, + index: u32, + pixel_format: u32, + ) -> IoctlResult { + if pixel_format != self.config.format.fourcc() { + return Err(libc::EINVAL); + } + if index > 0 { + return Err(libc::EINVAL); + } + + Ok(bindings::v4l2_frmsizeenum { + index, + pixel_format, + type_: bindings::v4l2_frmsizetypes_V4L2_FRMSIZE_TYPE_DISCRETE, + __bindgen_anon_1: bindings::v4l2_frmsizeenum__bindgen_ty_1 { + discrete: bindings::v4l2_frmsize_discrete { + width: self.config.input_width, + height: self.config.input_height, + }, + }, + ..Default::default() + }) + } + + fn enum_frameintervals( + &mut self, + _session: &Self::Session, + index: u32, + pixel_format: u32, + width: u32, + height: u32, + ) -> IoctlResult { + if pixel_format != self.config.format.fourcc() { + return Err(libc::EINVAL); + } + if width != self.config.input_width || height != self.config.input_height { + return Err(libc::EINVAL); + } + if index > 0 { + return Err(libc::EINVAL); + } + + let (numerator, denominator) = self.config.fps_interval; + + Ok(bindings::v4l2_frmivalenum { + index, + pixel_format, + width, + height, + type_: bindings::v4l2_frmivaltypes_V4L2_FRMIVAL_TYPE_DISCRETE, + __bindgen_anon_1: bindings::v4l2_frmivalenum__bindgen_ty_1 { + discrete: bindings::v4l2_fract { + numerator, + denominator, + }, + }, + ..Default::default() + }) + } +} diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/main.rs b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/main.rs new file mode 100644 index 00000000000..fe5628c5c1b --- /dev/null +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/main.rs @@ -0,0 +1,142 @@ +// Copyright 2026, The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::path::PathBuf; +use std::sync::{Arc, RwLock}; + +use clap::Parser; +use vhost_user_backend::VhostUserDaemon; +use vhu_media::VhuMediaBackend; +use vhu_media::cli::Error; +use virtio_media::protocol::VirtioMediaDeviceConfig; +use virtio_media::v4l2r::ioctl::Capabilities; +use vm_memory::{GuestMemoryAtomic, GuestMemoryMmap}; + +mod device; +mod worker; + +type Result = std::result::Result; + +#[derive(Parser, Debug)] +#[clap(author, version, about, long_about = None)] +struct CmdLineArgs { + /// Location of vhost-user Unix domain socket. + #[clap(short, long, value_name = "SOCKET")] + socket_path: PathBuf, + /// Log verbosity, one of Off, Error, Warning, Info, Debug, Trace. + #[clap(short, long, default_value_t = log::LevelFilter::Debug)] + verbosity: log::LevelFilter, + /// Path to the host named pipe (FIFO). + #[clap(long = "input_path", value_name = "INPUT_PATH")] + input_path: PathBuf, + /// Width of the video stream in pixels. + #[clap(long = "input_width", value_name = "INPUT_WIDTH")] + input_width: u32, + /// Height of the video stream in pixels. + #[clap(long = "input_height", value_name = "INPUT_HEIGHT")] + input_height: u32, + /// Frames per second (e.g. 30 or 30000/1001). + #[clap(long = "input_fps", value_name = "INPUT_FPS")] + input_fps: String, +} + +fn parse_fps_to_interval(fps_str: &str) -> Option<(u32, u32)> { + if let Ok(fps) = fps_str.parse::() { + return Some((1, fps)); + } + let parts: Vec<&str> = fps_str.split('/').collect(); + if parts.len() == 2 { + if let (Ok(num), Ok(den)) = (parts[0].parse::(), parts[1].parse::()) { + return Some((den, num)); + } + } + None +} + +#[derive(Clone, Debug)] +pub struct Config { + pub socket_path: PathBuf, + pub input_path: PathBuf, + pub input_width: u32, + pub input_height: u32, + pub fps_interval: (u32, u32), + pub format: device::Format, +} + +impl TryFrom for Config { + type Error = Error; + + fn try_from(args: CmdLineArgs) -> Result { + let fps_interval = parse_fps_to_interval(&args.input_fps) + .ok_or_else(|| Error::InvalidArgument(format!("Invalid FPS format: {}", args.input_fps)))?; + Ok(Config { + socket_path: args.socket_path, + input_path: args.input_path, + input_width: args.input_width, + input_height: args.input_height, + fps_interval, + format: device::Format::Yuv420M, + }) + } +} + +fn init_logging(verbosity: log::LevelFilter) -> Result<()> { + let mut builder = env_logger::Builder::new(); + builder.filter_level(verbosity); + builder.format_timestamp_secs(); + builder.init(); + Ok(()) +} + +const VFL_TYPE_VIDEO: u32 = 0; + +fn start_backend(config: Config) -> Result<()> { + let socket_path = config.socket_path.clone(); + let mut card = [0u8; 32]; + let card_name = "v4l2_stream_proxy"; + card[0..card_name.len()].copy_from_slice(card_name.as_bytes()); + + loop { + let caps = Capabilities::VIDEO_CAPTURE_MPLANE | Capabilities::STREAMING; + let device_config = VirtioMediaDeviceConfig { + device_caps: caps.bits(), + device_type: VFL_TYPE_VIDEO, + card, + }; + let backend_config = config.clone(); + let backend = Arc::new(RwLock::new(VhuMediaBackend::new( + device_config, + move |event_queue, host_mapper| { + crate::device::V4l2Stream::new(event_queue, host_mapper, backend_config.clone()) + }, + ))); + let mut daemon = VhostUserDaemon::new( + String::from("vhost-user-media-backend"), + backend, + GuestMemoryAtomic::new(GuestMemoryMmap::new()), + ) + .map_err(Error::CouldNotCreateDaemon)?; + log::info!("vhost-user-media-backend daemon started"); + daemon.serve(&socket_path).map_err(Error::ServeFailed)?; + log::info!("vhost-user-media-backend daemon closed gracefully"); + } +} + +fn main() -> Result<()> { + let args = CmdLineArgs::parse(); + + init_logging(args.verbosity)?; + + start_backend(Config::try_from(args)?) +} diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/worker.rs b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/worker.rs new file mode 100644 index 00000000000..3f55fd189f1 --- /dev/null +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy/src/worker.rs @@ -0,0 +1,311 @@ +// Copyright 2026, The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::fs::File; +use std::io::{Read, Write, Seek, SeekFrom}; +use std::os::fd::AsFd; +use std::sync::{Arc, Mutex}; +use std::sync::mpsc::{Receiver, Sender}; +use std::thread::JoinHandle; +use std::time::Instant; +use std::os::unix::fs::OpenOptionsExt; + +use nix::sys::eventfd::EventFd; +use nix::poll::{poll, PollFd, PollFlags, PollTimeout}; +use virtio_media::VirtioMediaEventQueue; +use virtio_media::protocol::{DequeueBufferEvent, V4l2Event}; + +use crate::Config; +use crate::device::{BufferState, SessionState}; + +#[derive(Debug)] +pub(crate) enum WorkerCmd { + Stop, + BufferQueued, +} + +pub(crate) struct WorkerHandle { + pub(crate) tx: Sender, + pub(crate) event_fd: Arc, + pub(crate) join_handle: JoinHandle<()>, +} + +/// Explicit representation of the worker thread states. +enum WorkerState { + /// FIFO is closed. Trying to open it. + Unopened, + /// FIFO is open, but waiting for buffers. + Idle { + fifo_file: File, + }, + /// FIFO is open and streaming into a buffer. + Streaming { + fifo_file: File, + buffer_idx: usize, + plane_idx: usize, + plane_offset: usize, + }, + /// Terminal state. Thread should exit. + Stopped, +} + +/// Helper function to drain control channel and check for Stop command. +fn should_stop(rx: &Receiver, event_fd: &EventFd) -> bool { + // Read eventfd to clear the notification + if let Err(e) = event_fd.read() { + if e != nix::Error::EWOULDBLOCK { + log::error!("Failed to read eventfd: {:?}", e); + } + } + // Drain channel + let mut stop = false; + while let Ok(cmd) = rx.try_recv() { + if let WorkerCmd::Stop = cmd { + stop = true; + } + } + stop +} + +fn handle_unopened( + config: &Config, + session_state: &Mutex, + rx: &Receiver, + event_fd: &EventFd, +) -> WorkerState { + match std::fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_NONBLOCK) + .open(&config.input_path) + { + Ok(file) => { + log::info!("FIFO opened"); + let mut s_state = session_state.lock().unwrap(); + if let Some(buf_idx) = s_state.queued_buffers.pop_front() { + WorkerState::Streaming { + fifo_file: file, + buffer_idx: buf_idx, + plane_idx: 0, + plane_offset: 0, + } + } else { + WorkerState::Idle { fifo_file: file } + } + } + Err(e) => { + log::error!("Failed to open FIFO: {:?}", e); + if should_stop(rx, event_fd) { + log::info!("Worker thread stopped during open retry"); + WorkerState::Stopped + } else { + WorkerState::Unopened + } + } + } +} + +fn handle_idle( + fifo_file: File, + session_state: &Mutex, + rx: &Receiver, + event_fd: &EventFd, +) -> WorkerState { + let mut poll_fds = [PollFd::new(event_fd.as_fd(), PollFlags::POLLIN)]; + match poll(&mut poll_fds, PollTimeout::NONE) { + Ok(_) => {} + Err(e) if e == nix::Error::EINTR => return WorkerState::Idle { fifo_file }, + Err(e) => { + log::error!("Poll error in Idle: {:?}", e); + return WorkerState::Idle { fifo_file }; + } + } + + if poll_fds[0].revents().unwrap_or(PollFlags::empty()).contains(PollFlags::POLLIN) { + if should_stop(rx, event_fd) { + return WorkerState::Stopped; + } + let mut s_state = session_state.lock().unwrap(); + if let Some(buf_idx) = s_state.queued_buffers.pop_front() { + WorkerState::Streaming { + fifo_file, + buffer_idx: buf_idx, + plane_idx: 0, + plane_offset: 0, + } + } else { + WorkerState::Idle { fifo_file } + } + } else { + WorkerState::Idle { fifo_file } + } +} + +fn handle_streaming( + mut fifo_file: File, + buffer_idx: usize, + mut plane_idx: usize, + mut plane_offset: usize, + session_state: &Mutex, + rx: &Receiver, + event_fd: &EventFd, + plane_sizes: &[usize], + local_buf: &mut [u8], + evt_queue: &Mutex, +) -> WorkerState { + let mut poll_fds = [ + PollFd::new(event_fd.as_fd(), PollFlags::POLLIN), + PollFd::new(fifo_file.as_fd(), PollFlags::POLLIN), + ]; + match poll(&mut poll_fds, PollTimeout::NONE) { + Ok(_) => {} + Err(e) if e == nix::Error::EINTR => { + return WorkerState::Streaming { fifo_file, buffer_idx, plane_idx, plane_offset }; + } + Err(e) => { + log::error!("Poll error in Streaming: {:?}", e); + return WorkerState::Streaming { fifo_file, buffer_idx, plane_idx, plane_offset }; + } + } + + if poll_fds[0].revents().unwrap_or(PollFlags::empty()).contains(PollFlags::POLLIN) { + if should_stop(rx, event_fd) { + return WorkerState::Stopped; + } + } + + if poll_fds[1].revents().unwrap_or(PollFlags::empty()).contains(PollFlags::POLLIN) { + let plane_size = plane_sizes[plane_idx]; + let remaining = plane_size - plane_offset; + let read_chunk = std::cmp::min(local_buf.len(), remaining); + + match fifo_file.read(&mut local_buf[..read_chunk]) { + Ok(0) => { + log::info!("FIFO EOF, writer disconnected. Re-opening..."); + WorkerState::Unopened + } + Ok(bytes_read) => { + let mut s_state = session_state.lock().unwrap(); + let session_id = s_state.id; + let buffer = &mut s_state.buffers[buffer_idx]; + let plane = &mut buffer.planes[plane_idx]; + + if let Err(e) = plane.fd.as_file().seek(SeekFrom::Start(plane_offset as u64)) { + log::error!("Seek error: {:?}", e); + return WorkerState::Stopped; + } + + if let Err(e) = plane.fd.as_file().write_all(&local_buf[..bytes_read]) { + log::error!("Write error: {:?}", e); + return WorkerState::Stopped; + } + + plane_offset += bytes_read; + if plane_offset == plane_size { + plane_idx += 1; + plane_offset = 0; + + if plane_idx == plane_sizes.len() { + let sequence = s_state.sequence; + s_state.sequence += 1; + let now = Instant::now(); + let delta = now.duration_since(s_state.last_frame_time); + s_state.last_frame_time = now; + + log::info!("Frame completed: session {}, seq {}, buf_idx {}, delta {:?}", session_id, sequence, buffer_idx, delta); + + s_state.buffers[buffer_idx].set_state(BufferState::Outgoing { sequence }); + let v4l2_buf = s_state.buffers[buffer_idx].v4l2_buffer.clone(); + + evt_queue.lock().unwrap().send_event(V4l2Event::DequeueBuffer(DequeueBufferEvent::new( + session_id, + v4l2_buf, + ))); + + if let Some(next_buf_idx) = s_state.queued_buffers.pop_front() { + WorkerState::Streaming { + fifo_file, + buffer_idx: next_buf_idx, + plane_idx: 0, + plane_offset: 0, + } + } else { + WorkerState::Idle { fifo_file } + } + } else { + WorkerState::Streaming { fifo_file, buffer_idx, plane_idx, plane_offset } + } + } else { + WorkerState::Streaming { fifo_file, buffer_idx, plane_idx, plane_offset } + } + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + WorkerState::Streaming { fifo_file, buffer_idx, plane_idx, plane_offset } + } + Err(e) => { + log::error!("Read error: {:?}", e); + WorkerState::Unopened + } + } + } else { + WorkerState::Streaming { fifo_file, buffer_idx, plane_idx, plane_offset } + } +} + +pub(crate) fn worker_thread_loop( + config: Config, + evt_queue: Arc>, + session_state: Arc>, + rx: Receiver, + event_fd: Arc, +) { + log::info!("Worker thread started for FIFO: {:?}", config.input_path); + + let plane_sizes = config.format.plane_sizes(config.input_width, config.input_height); + let mut local_buf = [0u8; 4096]; + + let mut state = WorkerState::Unopened; + + while !matches!(state, WorkerState::Stopped) { + state = match state { + WorkerState::Unopened => { + handle_unopened(&config, &session_state, &rx, &event_fd) + } + WorkerState::Idle { fifo_file } => { + handle_idle(fifo_file, &session_state, &rx, &event_fd) + } + WorkerState::Streaming { + fifo_file, + buffer_idx, + plane_idx, + plane_offset, + } => { + handle_streaming( + fifo_file, + buffer_idx, + plane_idx, + plane_offset, + &session_state, + &rx, + &event_fd, + &plane_sizes, + &mut local_buf, + &*evt_queue, + ) + } + WorkerState::Stopped => WorkerState::Stopped, + }; + } + + log::info!("Worker thread stopped"); +} diff --git a/base/cvd/cuttlefish/host/commands/vhost_user_media/vhu_media/src/lib.rs b/base/cvd/cuttlefish/host/commands/vhost_user_media/vhu_media/src/lib.rs index 49b20714669..d476c4321ac 100644 --- a/base/cvd/cuttlefish/host/commands/vhost_user_media/vhu_media/src/lib.rs +++ b/base/cvd/cuttlefish/host/commands/vhost_user_media/vhu_media/src/lib.rs @@ -178,25 +178,22 @@ impl EventQueue { fn send_events(&mut self, event: V4l2Event) -> Result<(), VhuMediaBackendError> { let vring = self.vring.clone(); let mem = self.mem.clone(); - let requests: Vec<_> = vring.get_mut().get_queue_mut().iter(mem)?.collect(); - if requests.is_empty() { - return Err(VhuMediaBackendError::DescriptorUnavailable); - } - for desc_chain in requests { - let mem = self.mem.clone(); - let head_index = desc_chain.head_index(); - let mut writer = desc_chain.writer(&mem)?; - match event { - V4l2Event::DequeueBuffer(e) => WriteToDescriptorChain::write_obj(&mut writer, e)?, - V4l2Event::Error(e) => WriteToDescriptorChain::write_obj(&mut writer, e)?, - V4l2Event::Event(e) => WriteToDescriptorChain::write_obj(&mut writer, e)?, - } - vring - .get_mut() - .add_used(head_index, writer.bytes_written() as u32)?; - vring.signal_used_queue()?; - break; + let desc_chain = match vring.get_mut().get_queue_mut().iter(mem)?.next() { + Some(d) => d, + None => return Err(VhuMediaBackendError::DescriptorUnavailable), + }; + let mem = self.mem.clone(); + let head_index = desc_chain.head_index(); + let mut writer = desc_chain.writer(&mem)?; + match event { + V4l2Event::DequeueBuffer(e) => WriteToDescriptorChain::write_obj(&mut writer, e)?, + V4l2Event::Error(e) => WriteToDescriptorChain::write_obj(&mut writer, e)?, + V4l2Event::Event(e) => WriteToDescriptorChain::write_obj(&mut writer, e)?, } + vring + .get_mut() + .add_used(head_index, writer.bytes_written() as u32)?; + vring.signal_used_queue()?; Ok(()) } } diff --git a/base/cvd/cuttlefish/host/libs/config/cuttlefish_config.h b/base/cvd/cuttlefish/host/libs/config/cuttlefish_config.h index 7fda4b33569..34702367316 100644 --- a/base/cvd/cuttlefish/host/libs/config/cuttlefish_config.h +++ b/base/cvd/cuttlefish/host/libs/config/cuttlefish_config.h @@ -118,11 +118,20 @@ class CuttlefishConfig { kV4l2EmulatedCameraSPlane, kV4l2EmulatedCameraMPlane, kV4l2Proxy, + kV4l2StreamProxy, }; struct MediaConfig { MediaType type; std::string lens_facing; + + struct V4l2StreamProxyConfig { + std::string input_path; + int input_width; + int input_height; + std::string input_fps; + }; + std::optional v4l2_stream_proxy; }; void set_secure_hals(const std::set&); diff --git a/base/cvd/cuttlefish/host/libs/config/cuttlefish_config_instance.cpp b/base/cvd/cuttlefish/host/libs/config/cuttlefish_config_instance.cpp index 6dc22a1b4d6..02925134d39 100644 --- a/base/cvd/cuttlefish/host/libs/config/cuttlefish_config_instance.cpp +++ b/base/cvd/cuttlefish/host/libs/config/cuttlefish_config_instance.cpp @@ -1988,6 +1988,11 @@ CuttlefishConfig::InstanceSpecific::audio_settings() const { static constexpr char kMediaConfigs[] = "media_configs"; static constexpr char kMediaType[] = "type"; static constexpr char kMediaLensFacing[] = "lens_facing"; +static constexpr char kMediaInputPath[] = "input_path"; +static constexpr char kMediaInputWidth[] = "input_width"; +static constexpr char kMediaInputHeight[] = "input_height"; +static constexpr char kMediaInputFps[] = "input_fps"; + std::vector CuttlefishConfig::InstanceSpecific::media_configs() const { std::vector configs; @@ -1998,6 +2003,22 @@ CuttlefishConfig::InstanceSpecific::media_configs() const { if (json.isMember(kMediaLensFacing)) { config.lens_facing = json[kMediaLensFacing].asString(); } + if (config.type == CuttlefishConfig::MediaType::kV4l2StreamProxy) { + CuttlefishConfig::MediaConfig::V4l2StreamProxyConfig stream_config = {}; + if (json.isMember(kMediaInputPath)) { + stream_config.input_path = json[kMediaInputPath].asString(); + } + if (json.isMember(kMediaInputWidth)) { + stream_config.input_width = json[kMediaInputWidth].asInt(); + } + if (json.isMember(kMediaInputHeight)) { + stream_config.input_height = json[kMediaInputHeight].asInt(); + } + if (json.isMember(kMediaInputFps)) { + stream_config.input_fps = json[kMediaInputFps].asString(); + } + config.v4l2_stream_proxy = stream_config; + } configs.emplace_back(config); } return configs; @@ -2011,6 +2032,13 @@ void CuttlefishConfig::MutableInstanceSpecific::set_media_configs( Json::Value json(Json::objectValue); json[kMediaType] = static_cast(config.type); json[kMediaLensFacing] = config.lens_facing; + if (config.type == CuttlefishConfig::MediaType::kV4l2StreamProxy && + config.v4l2_stream_proxy.has_value()) { + json[kMediaInputPath] = config.v4l2_stream_proxy->input_path; + json[kMediaInputWidth] = config.v4l2_stream_proxy->input_width; + json[kMediaInputHeight] = config.v4l2_stream_proxy->input_height; + json[kMediaInputFps] = config.v4l2_stream_proxy->input_fps; + } configs_json.append(json); } diff --git a/base/cvd/cuttlefish/host/libs/config/known_paths.cpp b/base/cvd/cuttlefish/host/libs/config/known_paths.cpp index 9cda39ee53f..9bb71c20caa 100644 --- a/base/cvd/cuttlefish/host/libs/config/known_paths.cpp +++ b/base/cvd/cuttlefish/host/libs/config/known_paths.cpp @@ -227,4 +227,8 @@ std::string VhostUserMediaEmulatedCameraMPlaneBinary() { return HostBinaryPath("vhu_media_emulated_camera_mplane"); } +std::string VhostUserMediaV4l2StreamProxyBinary() { + return HostBinaryPath("vhu_media_v4l2_stream_proxy"); +} + } // namespace cuttlefish diff --git a/base/cvd/cuttlefish/host/libs/config/known_paths.h b/base/cvd/cuttlefish/host/libs/config/known_paths.h index 5095ad58b06..119655b4e00 100644 --- a/base/cvd/cuttlefish/host/libs/config/known_paths.h +++ b/base/cvd/cuttlefish/host/libs/config/known_paths.h @@ -70,6 +70,7 @@ std::string VhalProxyServerConfig(); std::string VhostUserInputBinary(); std::string VhostUserMediaEmulatedCameraSPlaneBinary(); std::string VhostUserMediaEmulatedCameraMPlaneBinary(); +std::string VhostUserMediaV4l2StreamProxyBinary(); std::string WebRtcBinary(); std::string WebRtcSigServerBinary(); std::string WebRtcSigServerProxyBinary(); diff --git a/base/cvd/cuttlefish/host/libs/config/media.cpp b/base/cvd/cuttlefish/host/libs/config/media.cpp index fa32eeede8b..35399febdea 100644 --- a/base/cvd/cuttlefish/host/libs/config/media.cpp +++ b/base/cvd/cuttlefish/host/libs/config/media.cpp @@ -16,6 +16,8 @@ #include "cuttlefish/host/libs/config/media.h" +#include + #include #include #include @@ -36,6 +38,7 @@ static constexpr char kMediaTypeV4l2EmulatedCameraSPlane[] = static constexpr char kMediaTypeV4l2EmulatedCameraMPlane[] = "v4l2_emulated_camera_mplane"; static constexpr char kMediaTypeV4l2Proxy[] = "v4l2_proxy"; +static constexpr char kMediaTypeV4l2Stream[] = "v4l2_stream_proxy"; Result> ParseMediaConfig( const std::string& flag) { @@ -50,6 +53,8 @@ Result> ParseMediaConfig( type = CuttlefishConfig::MediaType::kV4l2EmulatedCameraMPlane; } else if (type_str == kMediaTypeV4l2Proxy) { type = CuttlefishConfig::MediaType::kV4l2Proxy; + } else if (type_str == kMediaTypeV4l2Stream) { + type = CuttlefishConfig::MediaType::kV4l2StreamProxy; } else { return CF_ERRF("Unknown media type value: \"{}\"", type_str); } @@ -74,9 +79,46 @@ Result> ParseMediaConfig( "Invalid lens_facing value: " << lens_facing); } + std::optional + v4l2_stream_proxy; + if (type == CuttlefishConfig::MediaType::kV4l2StreamProxy) { + CuttlefishConfig::MediaConfig::V4l2StreamProxyConfig stream_config = {}; + + auto source_it = props.find("input_path"); + CF_EXPECT(source_it != props.end(), + "Missing 'input_path' for v4l2_stream_proxy"); + stream_config.input_path = source_it->second; + + auto width_it = props.find("input_width"); + CF_EXPECT(width_it != props.end(), + "Missing 'input_width' for v4l2_stream_proxy"); + CF_EXPECT( + android::base::ParseInt(width_it->second, &stream_config.input_width), + "Failed to parse input_width"); + CF_EXPECT(stream_config.input_width > 0, + "input_width must be positive: " << stream_config.input_width); + + auto height_it = props.find("input_height"); + CF_EXPECT(height_it != props.end(), + "Missing 'input_height' for v4l2_stream_proxy"); + CF_EXPECT( + android::base::ParseInt(height_it->second, &stream_config.input_height), + "Failed to parse input_height"); + CF_EXPECT(stream_config.input_height > 0, + "input_height must be positive: " << stream_config.input_height); + + auto fps_it = props.find("input_fps"); + CF_EXPECT(fps_it != props.end(), + "Missing 'input_fps' for v4l2_stream_proxy"); + stream_config.input_fps = fps_it->second; + + v4l2_stream_proxy = stream_config; + } + return CuttlefishConfig::MediaConfig{ .type = type, .lens_facing = lens_facing, + .v4l2_stream_proxy = v4l2_stream_proxy, }; } diff --git a/base/cvd/cuttlefish/host/libs/config/media.h b/base/cvd/cuttlefish/host/libs/config/media.h index 5276577b629..c42e21cc139 100644 --- a/base/cvd/cuttlefish/host/libs/config/media.h +++ b/base/cvd/cuttlefish/host/libs/config/media.h @@ -32,10 +32,20 @@ constexpr const char kMediaHelp[] = " 'v4l2_emulated_camera_mplane': emulated media capture device " "(multi-plane)\n" " 'v4l2_proxy': proxy a host V4L2 device into the guest\n" + " 'v4l2_stream_proxy': stream video from a host named pipe into the " + "guest\n\n" + "v4l2_stream_proxy properties:\n" + " 'input_path': path to the host named pipe\n" + " 'input_width': width of the video stream in pixels\n" + " 'input_height': height of the video stream in pixels\n" + " 'input_fps': frames per second (e.g., 30 or 30000/1001)\n\n" "Supported keys:\n" - " 'lens_facing': optional, supported values: 'FRONT', 'BACK', 'EXTERNAL'\n" + " 'lens_facing': optional, supported values: 'FRONT', 'BACK', " + "'EXTERNAL'\n\n" "Example usage:\n" - " --media=v4l2_emulated_camera_mplane:lens_facing=BACK\n"; + " --media=v4l2_emulated_camera_mplane:lens_facing=BACK\n" + " --media=v4l2_stream_proxy:input_path=/tmp/fifo:" + "input_width=640:input_height=480:input_fps=30\n"; Result> ParseMediaConfig( const std::string& flag); diff --git a/base/cvd/cuttlefish/host/libs/vm_manager/crosvm_manager.cpp b/base/cvd/cuttlefish/host/libs/vm_manager/crosvm_manager.cpp index 16a1fb1b0ae..18fe168a3ca 100644 --- a/base/cvd/cuttlefish/host/libs/vm_manager/crosvm_manager.cpp +++ b/base/cvd/cuttlefish/host/libs/vm_manager/crosvm_manager.cpp @@ -975,7 +975,8 @@ Result> CrosvmManager::StartCommands( for (int index = 0; index < instance.media_configs().size(); index++) { auto config = instance.media_configs()[index]; if (config.type == CuttlefishConfig::MediaType::kV4l2EmulatedCameraSPlane || - config.type == CuttlefishConfig::MediaType::kV4l2EmulatedCameraMPlane) { + config.type == CuttlefishConfig::MediaType::kV4l2EmulatedCameraMPlane || + config.type == CuttlefishConfig::MediaType::kV4l2StreamProxy) { crosvm_cmd.Cmd().AddParameter("--vhost-user=type=media,socket=", instance.media_socket_path(index)); } else if (config.type == CuttlefishConfig::MediaType::kV4l2Proxy) { diff --git a/base/cvd/cuttlefish/package/BUILD.bazel b/base/cvd/cuttlefish/package/BUILD.bazel index 395c0ebf186..aa2460d12a0 100644 --- a/base/cvd/cuttlefish/package/BUILD.bazel +++ b/base/cvd/cuttlefish/package/BUILD.bazel @@ -99,6 +99,7 @@ package_files( "cuttlefish-common/bin/unpack_bootimg.py": "@mkbootimg//:unpack_bootimg.py", "cuttlefish-common/bin/vhu_media_emulated_camera_splane": "//cuttlefish/host/commands/vhost_user_media/emulated_camera_splane", "cuttlefish-common/bin/vhu_media_emulated_camera_mplane": "//cuttlefish/host/commands/vhost_user_media/emulated_camera_mplane", + "cuttlefish-common/bin/vhu_media_v4l2_stream_proxy": "//cuttlefish/host/commands/vhost_user_media/v4l2_stream_proxy", "cuttlefish-common/bin/vk_lavapipe_icd.json": "//cuttlefish/host/graphics/vulkan:vk_lavapipe_icd", "cuttlefish-common/bin/vk_swiftshader_icd.json": "//cuttlefish/host/graphics/vulkan:vk_swiftshader_icd", "cuttlefish-common/bin/webRTC": "//cuttlefish/host/frontend/webrtc:webRTC", diff --git a/e2etests/cvd/media_tests/v4l2_stream_proxy/v4l2compliance/BUILD.bazel b/e2etests/cvd/media_tests/v4l2_stream_proxy/v4l2compliance/BUILD.bazel new file mode 100644 index 00000000000..8ae8f33fabf --- /dev/null +++ b/e2etests/cvd/media_tests/v4l2_stream_proxy/v4l2compliance/BUILD.bazel @@ -0,0 +1,33 @@ +# Copyright (C) 2026 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +load("@rules_go//go:def.bzl", "go_test") + +go_test( + name = "media_tests", + size = "large", + srcs = ["main_test.go"], + data = ["//:debian_substitution_marker"], + env = {"LOCAL_DEBIAN_SUBSTITUTION_MARKER_FILE": "$(rlocationpath //:debian_substitution_marker)"}, + tags = [ + "exclusive", + "external", + "no-sandbox", + "requires_ab", + "supports-graceful-termination", + ], + deps = [ + "//cvd/common", + ], +) diff --git a/e2etests/cvd/media_tests/v4l2_stream_proxy/v4l2compliance/main_test.go b/e2etests/cvd/media_tests/v4l2_stream_proxy/v4l2compliance/main_test.go new file mode 100644 index 00000000000..669ddb608ec --- /dev/null +++ b/e2etests/cvd/media_tests/v4l2_stream_proxy/v4l2compliance/main_test.go @@ -0,0 +1,130 @@ +// Copyright (C) 2026 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "fmt" + "log" + "os" + "path/filepath" + "strings" + "syscall" + "testing" + "time" + + "github.com/google/android-cuttlefish/e2etests/cvd/common" +) + +func TestV4l2StreamProxyCompliance(t *testing.T) { + testcases := []struct { + branch string + target string + }{ + { + branch: "git_main", + target: "aosp_cf_x86_64_only_phone-trunk_staging-userdebug", + }, + { + branch: "git_main-throttled-nightly", + target: "aosp_cf_x86_64_auto-trunk_staging-userdebug", + }, + } + + for _, tc := range testcases { + t.Run(fmt.Sprintf("BUILD-%s/%s", tc.branch, tc.target), func(t *testing.T) { + c := e2etests.TestContext{} + c.SetUp(t) + defer c.TearDown() + + if _, err := c.CVDFetch(e2etests.FetchArgs{ + DefaultBuildBranch: tc.branch, + DefaultBuildTarget: tc.target, + }); err != nil { + t.Fatal(err) + } + + fifoPath := filepath.Join(t.TempDir(), "v4l2_fifo") + if err := syscall.Mkfifo(fifoPath, 0666); err != nil { + t.Fatalf("failed to create fifo %q: %v", fifoPath, err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go func() { + f, err := os.OpenFile(fifoPath, os.O_WRONLY, 0) + if err != nil { + log.Printf("failed to open fifo for writing: %v", err) + return + } + defer f.Close() + + frameSize := int(640 * 480 * 1.5) + buf := make([]byte, frameSize) + + ticker := time.NewTicker(33 * time.Millisecond) // ~30fps + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _, err := f.Write(buf) + if err != nil { + log.Printf("failed to write to fifo: %v", err) + return + } + } + } + }() + + mediaArg := fmt.Sprintf("--media=v4l2_stream_proxy:input_path=%s:input_width=640:input_height=480:input_fps=30", fifoPath) + if _, err := c.CVDCreate(e2etests.CreateArgs{ + Args: []string{mediaArg}, + }); err != nil { + t.Fatal(err) + } + + if err := c.RunAdbWaitForDevice(); err != nil { + t.Fatalf("failed to wait for Cuttlefish device to connect to adb: %w", err) + } + + // Find video node dynamically + videoNode := "" + for i := 0; i < 10; i++ { + node := fmt.Sprintf("/dev/video%d", i) + out, err := c.RunCmd("adb", "shell", "su", "0", "v4l2-ctl", "-d", node, "--info") + if err == nil && strings.Contains(out.Stdout, "v4l2_stream_proxy") { + videoNode = node + break + } + } + if videoNode == "" { + t.Fatal("v4l2_stream_proxy device not found in guest") + } + t.Logf("Found v4l2_stream_proxy device at %s", videoNode) + + if _, err := c.RunCmd("adb", "shell", "su", "0", "v4l2-ctl", "--list-devices"); err != nil { + t.Fatalf("v4l2-ctl --list-devices failed: %w", err) + } + + if _, err := c.RunCmd("adb", "shell", "su", "0", "v4l2-compliance", "-d", videoNode, "-s"); err != nil { + t.Fatalf("v4l2-compliance failed: %w", err) + } + }) + } +} diff --git a/e2etests/cvd/media_tests/v4l2compliance/main_test.go b/e2etests/cvd/media_tests/v4l2compliance/main_test.go index 3af9d484d08..9b80ada7c8e 100644 --- a/e2etests/cvd/media_tests/v4l2compliance/main_test.go +++ b/e2etests/cvd/media_tests/v4l2compliance/main_test.go @@ -73,3 +73,5 @@ func TestEmulatedCameraV4l2Compliance(t *testing.T) { }) } } + + diff --git a/tools/testutils/ffmpeg_v4l2_stream_proxy.sh b/tools/testutils/ffmpeg_v4l2_stream_proxy.sh new file mode 100755 index 00000000000..dc82e74f34a --- /dev/null +++ b/tools/testutils/ffmpeg_v4l2_stream_proxy.sh @@ -0,0 +1,120 @@ +#!/bin/bash + +# Copyright (C) 2026 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +set -e + +# Defaults +WIDTH=640 +HEIGHT=480 +FPS=30 +FIFO_PATH="/tmp/v4l2_fifo" +VIDEO_FILE="" + +usage() { + echo "Usage: $0 [-f ] [-w ] [-H ] [-r ] [-p ] [-h]" + echo " -f : Path to host video file to stream (e.g. MJPEG, MP4)" + echo " -w : Video width for testsrc (default: 640)" + echo " -H : Video height for testsrc (default: 480)" + echo " -r : Video FPS for testsrc (default: 30)" + echo " -p : Path to FIFO (default: /tmp/v4l2_fifo)" + echo " -h: Show this help message" + exit 0 +} + +while getopts "f:w:H:r:p:h" opt; do + case $opt in + f) VIDEO_FILE="$OPTARG" ;; + w) WIDTH="$OPTARG" ;; + H) HEIGHT="$OPTARG" ;; + r) FPS="$OPTARG" ;; + p) FIFO_PATH="$OPTARG" ;; + h) usage ;; + *) usage ;; + esac +done + +# Check dependencies +if ! command -v ffmpeg &>/dev/null; then + echo "Error: ffmpeg is not installed!" >&2 + exit 1 +fi + +if [ -n "$VIDEO_FILE" ] && ! command -v ffprobe &>/dev/null; then + echo "Error: ffprobe is not installed (required when -f is used)!" >&2 + exit 1 +fi + +FFMPEG_PID="" +CREATED_FIFO=false + +cleanup() { + echo "Stopping ffmpeg stream..." + if [ -n "$FFMPEG_PID" ]; then + echo "Killing ffmpeg (PID: $FFMPEG_PID)..." + kill -9 $FFMPEG_PID 2>/dev/null || true + wait $FFMPEG_PID 2>/dev/null || true + fi + if [ "$CREATED_FIFO" = true ] && [ -p "$FIFO_PATH" ]; then + echo "Removing created FIFO: $FIFO_PATH" + rm -f "$FIFO_PATH" + fi + exit 0 +} +# We use EXIT trap to ensure cleanup runs when the script exits (normally or via signal) +# We also trap INT and TERM explicitly to trigger exit (which triggers EXIT trap) +trap "exit 0" INT TERM +trap cleanup EXIT + +if [ ! -p "$FIFO_PATH" ]; then + echo "Creating FIFO at $FIFO_PATH" + mkfifo "$FIFO_PATH" + CREATED_FIFO=true +fi + +FFMPEG_ARGS=() + +if [ -n "$VIDEO_FILE" ]; then + if [ ! -f "$VIDEO_FILE" ]; then + echo "Error: Video file $VIDEO_FILE not found!" >&2 + exit 1 + fi + echo "Analyzing video file: $VIDEO_FILE..." + WIDTH=$(ffprobe -v error -select_streams v:0 -show_entries stream=width -of default=nw=1:nk=1 "$VIDEO_FILE") + HEIGHT=$(ffprobe -v error -select_streams v:0 -show_entries stream=height -of default=nw=1:nk=1 "$VIDEO_FILE") + FPS_RATIO=$(ffprobe -v error -select_streams v:0 -show_entries stream=r_frame_rate -of default=nw=1:nk=1 "$VIDEO_FILE") + if [ "$FPS_RATIO" = "0/0" ] || [ -z "$FPS_RATIO" ]; then + FPS_RATIO=$(ffprobe -v error -select_streams v:0 -show_entries stream=avg_frame_rate -of default=nw=1:nk=1 "$VIDEO_FILE") + fi + FPS="$FPS_RATIO" + echo "Detected parameters: Width=$WIDTH, Height=$HEIGHT, FPS=$FPS" + + FFMPEG_ARGS=(-re -stream_loop -1 -i "$VIDEO_FILE") + echo "Streaming from $VIDEO_FILE to $FIFO_PATH..." +else + FFMPEG_ARGS=(-re -f lavfi -i "testsrc=size=${WIDTH}x${HEIGHT}:rate=${FPS}") + echo "Streaming testsrc (${WIDTH}x${HEIGHT} @ ${FPS}fps) to $FIFO_PATH..." +fi + +echo "Suggested CVD config: --media=v4l2_stream_proxy:input_path=$FIFO_PATH:input_width=$WIDTH:input_height=$HEIGHT:input_fps=$FPS" +echo "Streaming started. Press Ctrl+C to stop." + +while true; do + echo "Starting ffmpeg..." + ffmpeg -y "${FFMPEG_ARGS[@]}" -f rawvideo -pix_fmt yuv420p "$FIFO_PATH" >/dev/null 2>&1 & + FFMPEG_PID=$! + wait $FFMPEG_PID || true + FFMPEG_PID="" +done