use std::collections::VecDeque;
const SW_REPLAY_FRAME_CAP: usize = 64;
use derive_more::{IsVariant, TryUnwrap, Unwrap};
use ffmpeg_next::{Packet, codec::Parameters, frame};
use mediadecode::{
Received, Sent, Timebase,
decoder::{ScaledOutputCapability, VideoStreamDecoder},
frame::VideoFrame,
packet::VideoPacket,
};
use crate::{
Backend, DecoderLimits, Error, Ffmpeg, Frame, VideoDecoder, boundary,
convert::{self, ConvertError},
decoder::{build_codec_context, try_clone_parameters},
error::FallbackFailed,
extras::{VideoFrameExtra, VideoPacketExtra},
frame::alloc_av_video_frame,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum DecodePath {
Auto,
AnyHardware,
Hardware(Backend),
Software,
}
pub struct CarrierVideoStreamDecoder<C: crate::FfmpegCarrier> {
state: DecodeState,
path: DecodePath,
parameters: Parameters,
hw_scratch: Frame,
sw_scratch: frame::Video,
sw_replay_frames: VecDeque<frame::Video>,
limits: DecoderLimits,
eof_sent: bool,
time_base: Timebase,
scratch_pending: bool,
_carrier: core::marker::PhantomData<C>,
}
pub(crate) trait HwInner: Send {
unsafe fn send_packet(&mut self, packet: &Packet) -> Result<Sent, Error>;
fn receive_frame(&mut self, frame: &mut Frame) -> Result<Received, Error>;
fn send_eof(&mut self) -> Result<Sent, Error>;
fn flush(&mut self) -> Result<(), Error>;
fn as_video_decoder(&self) -> Option<&VideoDecoder>;
fn scaled_output_capability(&self) -> ScaledOutputCapability {
ScaledOutputCapability::Unsupported
}
fn request_scaled_output(&mut self, size: (u32, u32)) -> ScaledOutputCapability {
let _ = size;
ScaledOutputCapability::Unsupported
}
fn cancel_scaled_output(&mut self) {}
}
impl HwInner for VideoDecoder {
#[inline]
unsafe fn send_packet(&mut self, packet: &Packet) -> Result<Sent, Error> {
unsafe { VideoDecoder::send_av_packet(self, packet) }
}
#[inline]
fn receive_frame(&mut self, frame: &mut Frame) -> Result<Received, Error> {
VideoDecoder::receive_frame(self, frame)
}
#[inline]
fn send_eof(&mut self) -> Result<Sent, Error> {
VideoDecoder::send_eof(self)
}
#[inline]
fn flush(&mut self) -> Result<(), Error> {
VideoDecoder::flush(self);
Ok(())
}
#[inline]
fn as_video_decoder(&self) -> Option<&VideoDecoder> {
Some(self)
}
#[inline]
fn scaled_output_capability(&self) -> ScaledOutputCapability {
VideoDecoder::scaled_output_capability(self)
}
#[inline]
fn request_scaled_output(&mut self, size: (u32, u32)) -> ScaledOutputCapability {
VideoDecoder::request_scaled_output(self, size)
}
#[inline]
fn cancel_scaled_output(&mut self) {
VideoDecoder::cancel_scaled_output(self);
}
}
enum DecodeState {
Hw(Box<dyn HwInner>),
Sw(SwDecoder),
}
pub(crate) struct SwDecoder {
decoder: ffmpeg_next::decoder::Video,
_callback_state: crate::decoder::InstalledCallbackState,
}
impl SwDecoder {
pub(crate) fn state(&self) -> *const crate::ffi::CallbackState {
self._callback_state.as_ptr()
}
}
impl core::ops::Deref for SwDecoder {
type Target = ffmpeg_next::decoder::Video;
fn deref(&self) -> &Self::Target {
&self.decoder
}
}
impl core::ops::DerefMut for SwDecoder {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.decoder
}
}
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierVideoStreamDecoder<C> {
pub(crate) fn open_impl(
parameters: Parameters,
time_base: Timebase,
limits: DecoderLimits,
) -> Result<Self, Error> {
Self::open_as_impl(parameters, time_base, limits, DecodePath::Auto)
}
pub(crate) fn open_as_impl(
parameters: Parameters,
time_base: Timebase,
limits: DecoderLimits,
path: DecodePath,
) -> Result<Self, Error> {
Self::open_as_in(
parameters,
time_base,
limits,
path,
crate::backend::probe_order(),
)
}
pub(crate) fn open_as_in(
parameters: Parameters,
time_base: Timebase,
limits: DecoderLimits,
path: DecodePath,
order: &[Backend],
) -> Result<Self, Error> {
let owned_parameters = try_clone_parameters(¶meters, limits.max_codec_parameter_bytes())?;
let hw_scratch = Frame::empty()?;
let sw_scratch = alloc_av_video_frame()?;
let state = match path {
DecodePath::Auto => match VideoDecoder::open_probing(
try_clone_parameters(&owned_parameters, limits.max_codec_parameter_bytes())?,
limits,
Some(time_base),
order,
) {
Ok(hw) => DecodeState::Hw(Box::new(hw)),
Err(Error::AllBackendsFailed(_)) => {
let sw = open_sw_decoder(&owned_parameters, limits, Some(time_base))?;
DecodeState::Sw(sw)
}
Err(other) => return Err(other),
},
DecodePath::AnyHardware => DecodeState::Hw(Box::new(VideoDecoder::open_probing(
try_clone_parameters(&owned_parameters, limits.max_codec_parameter_bytes())?,
limits,
Some(time_base),
order,
)?)),
DecodePath::Hardware(backend) => {
DecodeState::Hw(Box::new(VideoDecoder::open_with_limits_timed(
try_clone_parameters(&owned_parameters, limits.max_codec_parameter_bytes())?,
backend,
limits,
time_base,
)?))
}
DecodePath::Software => {
DecodeState::Sw(open_sw_decoder(&owned_parameters, limits, Some(time_base))?)
}
};
Ok(Self {
state,
path,
parameters: owned_parameters,
hw_scratch,
sw_scratch,
sw_replay_frames: VecDeque::new(),
eof_sent: false,
time_base,
limits,
scratch_pending: false,
_carrier: core::marker::PhantomData,
})
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) const fn is_software_impl(&self) -> bool {
matches!(self.state, DecodeState::Sw(_))
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) const fn is_hardware_impl(&self) -> bool {
matches!(self.state, DecodeState::Hw(_))
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) fn scaled_output_capability_impl(&self) -> ScaledOutputCapability {
match &self.state {
DecodeState::Hw(hw) => hw.scaled_output_capability(),
DecodeState::Sw(_) => ScaledOutputCapability::Unsupported,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) fn request_scaled_output_impl(&mut self, size: (u32, u32)) -> ScaledOutputCapability {
if self.scratch_pending {
tracing::debug!(
requested_width = size.0,
requested_height = size.1,
"mediadecode-ffmpeg: scaled-output request refused while a decoded frame is parked; \
the session returns to full size — drain it and ask again"
);
if let DecodeState::Hw(hw) = &mut self.state {
hw.cancel_scaled_output();
}
return ScaledOutputCapability::Unsupported;
}
match &mut self.state {
DecodeState::Hw(hw) => hw.request_scaled_output(size),
DecodeState::Sw(_) => ScaledOutputCapability::Unsupported,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) fn hardware_inner_impl(&self) -> Option<&VideoDecoder> {
match &self.state {
DecodeState::Hw(hw) => hw.as_video_decoder(),
DecodeState::Sw(_) => None,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) const fn time_base_impl(&self) -> Timebase {
self.time_base
}
#[cfg_attr(not(tarpaulin), inline(always))]
const fn may_open_software(&self) -> bool {
matches!(self.path, DecodePath::Auto | DecodePath::Software)
}
fn fall_back_to_sw(
&mut self,
unconsumed_packets: std::vec::Vec<ffmpeg_next::Packet>,
eof_pending: bool,
) -> Result<(), Error> {
tracing::info!(
packets_replayed = unconsumed_packets.len(),
eof_pending,
"mediadecode-ffmpeg: HW probe exhausted, falling back to software decode",
);
match self.fall_back_to_sw_inner(&unconsumed_packets, eof_pending) {
Ok(()) => Ok(()),
Err(source) => Err(Error::FallbackFailed(FallbackFailed::new(
Box::new(source),
unconsumed_packets,
))),
}
}
fn fall_back_to_sw_inner(
&mut self,
unconsumed_packets: &[ffmpeg_next::Packet],
eof_pending: bool,
) -> Result<(), Error> {
let mut sw = open_sw_decoder(&self.parameters, self.limits, Some(self.time_base))?;
let sw_state = sw.state();
let mut local_replay: VecDeque<frame::Video> = VecDeque::new();
unsafe fn drain_into(
sw: &mut ffmpeg_next::decoder::Video,
state: *const crate::ffi::CallbackState,
local_replay: &mut VecDeque<frame::Video>,
) -> std::result::Result<(), Error> {
loop {
let mut tmp = alloc_av_video_frame()?;
match sw.receive_frame(&mut tmp) {
Ok(()) => {
if local_replay.len() >= SW_REPLAY_FRAME_CAP {
tracing::error!(
cap = SW_REPLAY_FRAME_CAP,
"mediadecode-ffmpeg: SW fallback replay produced more frames than the \
replay cap allows; aborting fallback (no frames dropped — they're \
still in the SW decoder's internal queue and will be released when \
it drops)",
);
return Err(Error::Ffmpeg(ffmpeg_next::Error::Other {
errno: libc::ENOMEM,
}));
}
local_replay.push_back(tmp);
}
Err(ffmpeg_next::Error::Other { errno }) if errno == ffmpeg_next::error::EAGAIN => {
break;
}
Err(ffmpeg_next::Error::Eof) => break,
Err(other) => return Err(unsafe { crate::decoder::software_exit(state, other) }),
}
}
Ok(())
}
for pkt in unconsumed_packets {
let mut attempts: u32 = 0;
loop {
match sw.send_packet(pkt) {
Ok(()) => break,
Err(ffmpeg_next::Error::Other { errno }) if errno == ffmpeg_next::error::EAGAIN => {
unsafe { drain_into(&mut sw, sw_state, &mut local_replay) }?;
attempts += 1;
if attempts > 16 {
return Err(Error::Ffmpeg(ffmpeg_next::Error::Other {
errno: ffmpeg_next::error::EAGAIN,
}));
}
}
Err(other) => return Err(unsafe { crate::decoder::software_exit(sw_state, other) }),
}
}
}
if eof_pending {
let mut attempts: u32 = 0;
loop {
match sw.send_eof() {
Ok(()) => break,
Err(ffmpeg_next::Error::Other { errno }) if errno == ffmpeg_next::error::EAGAIN => {
unsafe { drain_into(&mut sw, sw_state, &mut local_replay) }?;
attempts += 1;
if attempts > 16 {
return Err(Error::Ffmpeg(ffmpeg_next::Error::Other {
errno: ffmpeg_next::error::EAGAIN,
}));
}
}
Err(other) => return Err(unsafe { crate::decoder::software_exit(sw_state, other) }),
}
}
}
unsafe { drain_into(&mut sw, sw_state, &mut local_replay) }?;
self.sw_replay_frames.append(&mut local_replay);
self.state = DecodeState::Sw(sw);
Ok(())
}
fn commit_delivery(
&mut self,
frame: VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
) {
self.scratch_pending = false;
*dst = frame;
}
const fn phase(&self) -> crate::decoder::SessionPhase {
if self.eof_sent {
crate::decoder::SessionPhase::Draining
} else {
crate::decoder::SessionPhase::Streaming
}
}
fn deliver_frame(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
) -> Result<Received, VideoDecodeError> {
let av_frame = match &mut self.state {
DecodeState::Hw(_) => unsafe { self.hw_scratch.as_inner_mut().as_ptr() },
DecodeState::Sw(_) => unsafe { self.sw_scratch.as_ptr() },
};
let converted = unsafe {
convert::av_frame_to_video_frame_as::<C>(av_frame, self.time_base, self.limits.frame())
};
match converted {
Ok(new_frame) => {
self.commit_delivery(new_frame, dst);
Ok(Received::Produced)
}
Err(e) => {
self.scratch_pending = crate::decoder::parks_in_decode(&e);
Err(VideoDecodeError::Convert(e))
}
}
}
}
#[cfg(test)]
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierVideoStreamDecoder<C> {
pub(crate) fn from_hw_inner_for_test(
hw: Box<dyn HwInner>,
parameters: Parameters,
time_base: Timebase,
) -> Result<Self, Error> {
Self::from_hw_inner_for_test_as(hw, parameters, time_base, DecodePath::Auto)
}
pub(crate) fn from_hw_inner_for_test_as(
hw: Box<dyn HwInner>,
parameters: Parameters,
time_base: Timebase,
path: DecodePath,
) -> Result<Self, Error> {
let limits = DecoderLimits::default();
let owned_parameters = try_clone_parameters(¶meters, limits.max_codec_parameter_bytes())?;
Ok(Self {
state: DecodeState::Hw(hw),
path,
parameters: owned_parameters,
hw_scratch: Frame::empty()?,
sw_scratch: alloc_av_video_frame()?,
sw_replay_frames: VecDeque::new(),
eof_sent: false,
time_base,
limits,
scratch_pending: false,
_carrier: core::marker::PhantomData,
})
}
pub(crate) const fn eof_sent_for_test(&self) -> bool {
self.eof_sent
}
pub(crate) fn sw_replay_frames_is_empty_for_test(&self) -> bool {
self.sw_replay_frames.is_empty()
}
}
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierVideoStreamDecoder<C> {
fn after_eof() -> VideoDecodeError {
VideoDecodeError::Decode(Error::Ffmpeg(ffmpeg_next::Error::Eof))
}
pub(crate) fn send_packet_impl(
&mut self,
packet: &VideoPacket<VideoPacketExtra, C::Buffer>,
) -> Result<Sent, VideoDecodeError> {
if !self.phase().accepts_input() {
return Err(Self::after_eof());
}
if self.scratch_pending {
return Ok(Sent::MustDrain);
}
let phase = self.phase();
let limits = self.limits.packet_limits();
let route = crate::BodyRoute::Submission;
let submit = |av_pkt: &Packet| {
match &mut self.state {
DecodeState::Hw(hw) => match unsafe { hw.send_packet(av_pkt) } {
Ok(status) => Ok(status),
Err(Error::AllBackendsFailed(p)) => {
if !self.may_open_software() {
return Err(VideoDecodeError::Decode(Error::AllBackendsFailed(p)));
}
let rescued = p.into_unconsumed_packets();
let eof_pending = self.eof_sent;
self
.fall_back_to_sw(rescued, eof_pending)
.map_err(VideoDecodeError::Decode)?;
if let DecodeState::Sw(sw) = &mut self.state {
let st = sw.state();
if let Err(e) = sw.send_packet(av_pkt) {
return unsafe { crate::decoder::software_send(st, e, phase) }
.map_err(VideoDecodeError::Decode);
}
}
Ok(Sent::Accepted)
}
Err(other) => Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
let st = sw.state();
match sw.send_packet(av_pkt) {
Ok(()) => Ok(Sent::Accepted),
Err(e) => unsafe { crate::decoder::software_send(st, e, phase) }
.map_err(VideoDecodeError::Decode),
}
}
}
};
unsafe { boundary::with_ffmpeg_video_packet::<C, _>(packet, limits, route, submit) }
.map_err(|e| VideoDecodeError::Decode(Error::PacketBuild(e)))?
}
pub(crate) fn receive_frame_impl(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
) -> Result<Received, VideoDecodeError> {
if let Some(replayed) = self.sw_replay_frames.front() {
let converted = unsafe {
convert::av_frame_to_video_frame_as::<C>(
replayed.as_ptr(),
self.time_base,
self.limits.frame(),
)
};
let new_frame = match converted {
Ok(new_frame) => new_frame,
Err(e) if crate::decoder::parks_in_decode(&e) => return Err(VideoDecodeError::Convert(e)),
Err(e) => {
self.sw_replay_frames.pop_front();
return Err(VideoDecodeError::Convert(e));
}
};
self.sw_replay_frames.pop_front();
self.commit_delivery(new_frame, dst);
return Ok(Received::Produced);
}
if self.scratch_pending {
return self.deliver_frame(dst);
}
let phase = self.phase();
loop {
match &mut self.state {
DecodeState::Hw(hw) => match hw.receive_frame(&mut self.hw_scratch) {
Ok(Received::Produced) => {
self.scratch_pending = true;
return self.deliver_frame(dst);
}
Ok(status) => return Ok(status),
Err(Error::AllBackendsFailed(p)) => {
if !self.may_open_software() {
return Err(VideoDecodeError::Decode(Error::AllBackendsFailed(p)));
}
let rescued = p.into_unconsumed_packets();
let eof_pending = self.eof_sent;
self
.fall_back_to_sw(rescued, eof_pending)
.map_err(VideoDecodeError::Decode)?;
if let Some(replayed) = self.sw_replay_frames.front() {
let converted = unsafe {
convert::av_frame_to_video_frame_as::<C>(
replayed.as_ptr(),
self.time_base,
self.limits.frame(),
)
};
let new_frame = match converted {
Ok(new_frame) => new_frame,
Err(e) if crate::decoder::parks_in_decode(&e) => {
return Err(VideoDecodeError::Convert(e));
}
Err(e) => {
self.sw_replay_frames.pop_front();
return Err(VideoDecodeError::Convert(e));
}
};
self.sw_replay_frames.pop_front();
self.commit_delivery(new_frame, dst);
return Ok(Received::Produced);
}
}
Err(other) => return Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
let st = sw.state();
match sw.receive_frame(&mut self.sw_scratch) {
Ok(()) => {
self.scratch_pending = true;
let converted = unsafe {
convert::av_frame_to_video_frame_as::<C>(
self.sw_scratch.as_ptr(),
self.time_base,
self.limits.frame(),
)
};
let new_frame = match converted {
Ok(new_frame) => new_frame,
Err(e) => {
self.scratch_pending = crate::decoder::parks_in_decode(&e);
return Err(VideoDecodeError::Convert(e));
}
};
self.commit_delivery(new_frame, dst);
return Ok(Received::Produced);
}
Err(e) => {
return unsafe { crate::decoder::software_receive(st, e, phase) }
.map_err(VideoDecodeError::Decode);
}
}
}
}
}
}
pub(crate) fn send_eof_impl(&mut self) -> Result<Sent, VideoDecodeError> {
if !self.phase().accepts_input() {
return Err(Self::after_eof());
}
if self.scratch_pending {
return Ok(Sent::MustDrain);
}
let phase = self.phase();
let outcome = match &mut self.state {
DecodeState::Hw(hw) => match hw.send_eof() {
Ok(status) => Ok(status),
Err(Error::AllBackendsFailed(p)) => {
if !self.may_open_software() {
return Err(VideoDecodeError::Decode(Error::AllBackendsFailed(p)));
}
let rescued = p.into_unconsumed_packets();
self
.fall_back_to_sw(rescued, true)
.map(|()| Sent::Accepted)
.map_err(VideoDecodeError::Decode)
}
Err(other) => Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
let st = sw.state();
match sw.send_eof() {
Ok(()) => Ok(Sent::Accepted),
Err(e) => {
unsafe { crate::decoder::software_send(st, e, phase) }.map_err(VideoDecodeError::Decode)
}
}
}
};
if matches!(outcome, Ok(Sent::Accepted)) {
self.eof_sent = true;
}
outcome
}
pub(crate) fn flush_impl(&mut self) -> Result<(), VideoDecodeError> {
self.sw_replay_frames.clear();
self.scratch_pending = false;
self.eof_sent = false;
match &mut self.state {
DecodeState::Hw(hw) => hw.flush().map_err(VideoDecodeError::Decode)?,
DecodeState::Sw(sw) => sw.flush(),
}
Ok(())
}
}
macro_rules! video_lane_face {
($($lane:ty),+ $(,)?) => { $(
impl CarrierVideoStreamDecoder<$lane> {
pub fn open(
parameters: Parameters,
time_base: Timebase,
limits: DecoderLimits,
) -> Result<Self, Error> {
Self::open_impl(parameters, time_base, limits)
}
pub fn open_as(
parameters: Parameters,
time_base: Timebase,
limits: DecoderLimits,
path: DecodePath,
) -> Result<Self, Error> {
Self::open_as_impl(parameters, time_base, limits, path)
}
pub const fn is_software(&self) -> bool {
self.is_software_impl()
}
pub const fn is_hardware(&self) -> bool {
self.is_hardware_impl()
}
pub fn scaled_output_capability(&self) -> ScaledOutputCapability {
self.scaled_output_capability_impl()
}
pub fn request_scaled_output(&mut self, size: (u32, u32)) -> ScaledOutputCapability {
self.request_scaled_output_impl(size)
}
pub fn hardware_inner(&self) -> Option<&VideoDecoder> {
self.hardware_inner_impl()
}
pub const fn time_base(&self) -> Timebase {
self.time_base_impl()
}
}
impl VideoStreamDecoder for CarrierVideoStreamDecoder<$lane> {
type Adapter = Ffmpeg;
type Buffer = <$lane as crate::FfmpegCarrier>::Buffer;
type Error = VideoDecodeError;
fn send_packet(
&mut self,
packet: &VideoPacket<VideoPacketExtra, Self::Buffer>,
) -> Result<Sent, Self::Error> {
self.send_packet_impl(packet)
}
fn receive_frame(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, Self::Buffer>,
) -> Result<Received, Self::Error> {
self.receive_frame_impl(dst)
}
fn send_eof(&mut self) -> Result<Sent, Self::Error> {
self.send_eof_impl()
}
fn flush(&mut self) -> Result<(), Self::Error> {
self.flush_impl()
}
fn scaled_output_capability(&self) -> ScaledOutputCapability {
self.scaled_output_capability_impl()
}
fn request_scaled_output(&mut self, size: (u32, u32)) -> ScaledOutputCapability {
self.request_scaled_output_impl(size)
}
}
)+ };
}
video_lane_face!(crate::View, crate::Owned);
fn open_sw_decoder(
parameters: &Parameters,
limits: DecoderLimits,
pkt_timebase: Option<Timebase>,
) -> Result<SwDecoder, Error> {
let (ctx, callback_state) = build_codec_context(parameters, limits, pkt_timebase)?;
let codec = crate::decoder::find_decoder(parameters)?;
let opened = ctx.decoder().open_as(codec).map_err(Error::Ffmpeg)?;
crate::decoder::ensure_video_codec_type(&opened)?;
Ok(SwDecoder {
decoder: ffmpeg_next::decoder::Video(opened),
_callback_state: callback_state,
})
}
#[derive(thiserror::Error, Debug, IsVariant, Unwrap, TryUnwrap)]
#[unwrap(ref, ref_mut)]
#[try_unwrap(ref, ref_mut)]
#[non_exhaustive]
pub enum VideoDecodeError {
#[error(transparent)]
Decode(#[from] Error),
#[error(transparent)]
Convert(#[from] ConvertError),
}
#[cfg(test)]
mod tests;