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::{Timebase, decoder::VideoStreamDecoder, frame::VideoFrame, packet::VideoPacket};
use crate::{
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,
};
pub struct CarrierVideoStreamDecoder<C: crate::FfmpegCarrier> {
state: DecodeState,
parameters: Parameters,
hw_scratch: Frame,
sw_scratch: frame::Video,
sw_replay_frames: VecDeque<frame::Video>,
limits: DecoderLimits,
eof_sent: bool,
degraded_resync_pending: bool,
degraded_keyframe_seen: bool,
degraded_packets_since_fallback: u64,
time_base: Timebase,
scratch_pending: bool,
_carrier: core::marker::PhantomData<C>,
}
pub(crate) trait HwInner: Send {
fn send_packet(&mut self, packet: &Packet) -> Result<(), Error>;
fn receive_frame(&mut self, frame: &mut Frame) -> Result<(), Error>;
fn send_eof(&mut self) -> Result<(), Error>;
fn flush(&mut self) -> Result<(), Error>;
fn as_video_decoder(&self) -> Option<&VideoDecoder>;
fn records_submissions(&self) -> bool;
}
impl HwInner for VideoDecoder {
#[inline]
fn records_submissions(&self) -> bool {
self.is_probing()
}
#[inline]
fn send_packet(&mut self, packet: &Packet) -> Result<(), Error> {
VideoDecoder::send_packet(self, packet)
}
#[inline]
fn receive_frame(&mut self, frame: &mut Frame) -> Result<(), Error> {
VideoDecoder::receive_frame(self, frame)
}
#[inline]
fn send_eof(&mut self) -> Result<(), 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)
}
}
enum DecodeState {
Hw(Box<dyn HwInner>),
Sw(SwDecoder),
}
pub(crate) struct SwDecoder {
decoder: ffmpeg_next::decoder::Video,
_callback_state: Box<crate::ffi::CallbackState>,
}
impl SwDecoder {
pub(crate) fn state(&self) -> *const crate::ffi::CallbackState {
&*self._callback_state
}
}
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
}
}
enum PostCommitInput<'a> {
Packet(&'a Packet),
FrameTime,
Eof,
}
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierVideoStreamDecoder<C> {
pub(crate) fn open_impl(
parameters: Parameters,
time_base: Timebase,
limits: DecoderLimits,
) -> 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 VideoDecoder::open_with_frame_limits(
try_clone_parameters(&owned_parameters, limits.max_codec_parameter_bytes())?,
limits,
) {
Ok(hw) => DecodeState::Hw(Box::new(hw)),
Err(Error::AllBackendsFailed(_)) => {
let sw = open_sw_decoder(&owned_parameters, limits)?;
DecodeState::Sw(sw)
}
Err(other) => return Err(other),
};
Ok(Self {
state,
parameters: owned_parameters,
hw_scratch,
sw_scratch,
sw_replay_frames: VecDeque::new(),
eof_sent: false,
degraded_resync_pending: false,
degraded_keyframe_seen: false,
degraded_packets_since_fallback: 0,
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 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
}
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)?;
let sw_state = sw.state();
let mut local_replay: VecDeque<frame::Video> = VecDeque::new();
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(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 => {
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(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 => {
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(crate::decoder::software_exit(sw_state, other)),
}
}
}
drain_into(&mut sw, sw_state, &mut local_replay)?;
self.sw_replay_frames.append(&mut local_replay);
self.state = DecodeState::Sw(sw);
Ok(())
}
fn degrade_to_sw(&mut self, input: PostCommitInput<'_>) -> Result<(), Error> {
match self.degrade_to_sw_inner(input) {
Ok(()) => Ok(()),
Err(budget @ Error::FrameBudgetExceeded(_)) => Err(budget),
Err(source) => Err(Error::FallbackFailed(FallbackFailed::new(
Box::new(source),
std::vec::Vec::new(),
))),
}
}
fn degrade_to_sw_inner(&mut self, input: PostCommitInput<'_>) -> Result<(), Error> {
let mut sw = open_sw_decoder(&self.parameters, self.limits)?;
let state = sw.state();
let mut forwarded_keyframe = false;
let mut forwarded_packet = false;
match input {
PostCommitInput::Packet(pkt) => {
sw.send_packet(pkt)
.map_err(|e| crate::decoder::software_exit(state, e))?;
forwarded_keyframe = pkt.is_key();
forwarded_packet = true;
}
PostCommitInput::FrameTime => {}
PostCommitInput::Eof => {
sw.send_eof()
.map_err(|e| crate::decoder::software_exit(state, e))?;
}
}
self.state = DecodeState::Sw(sw);
self.enter_degraded_resync();
if forwarded_keyframe {
self.note_degraded_keyframe(true);
}
if forwarded_packet {
self.count_degraded_packet();
}
Ok(())
}
#[inline]
fn enter_degraded_resync(&mut self) {
self.degraded_resync_pending = true;
self.degraded_keyframe_seen = false;
self.degraded_packets_since_fallback = 0;
}
#[inline]
fn note_degraded_keyframe(&mut self, is_key: bool) {
if self.degraded_resync_pending && is_key {
self.degraded_keyframe_seen = true;
}
}
#[inline]
fn count_degraded_packet(&mut self) {
if self.degraded_resync_pending {
self.degraded_packets_since_fallback = self.degraded_packets_since_fallback.saturating_add(1);
}
}
#[inline]
fn resync_on_frame(&mut self) {
if self.degraded_resync_pending && self.degraded_keyframe_seen {
self.clear_degraded_resync();
}
}
#[inline]
fn clear_degraded_resync(&mut self) {
self.degraded_resync_pending = false;
self.degraded_keyframe_seen = false;
self.degraded_packets_since_fallback = 0;
}
fn commit_delivery(
&mut self,
frame: VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
) {
self.scratch_pending = false;
self.resync_on_frame();
*dst = frame;
}
fn deliver_frame(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, C::Buffer>,
) -> Result<(), 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(())
}
Err(e) => {
self.scratch_pending = e.parks_in_decode();
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> {
let limits = DecoderLimits::default();
let owned_parameters = try_clone_parameters(¶meters, limits.max_codec_parameter_bytes())?;
Ok(Self {
state: DecodeState::Hw(hw),
parameters: owned_parameters,
hw_scratch: Frame::empty()?,
sw_scratch: alloc_av_video_frame()?,
sw_replay_frames: VecDeque::new(),
eof_sent: false,
degraded_resync_pending: false,
degraded_keyframe_seen: false,
degraded_packets_since_fallback: 0,
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) const fn degraded_resync_pending_for_test(&self) -> bool {
self.degraded_resync_pending
}
pub(crate) const fn degraded_keyframe_seen_for_test(&self) -> bool {
self.degraded_keyframe_seen
}
pub(crate) fn sw_replay_frames_is_empty_for_test(&self) -> bool {
self.sw_replay_frames.is_empty()
}
pub(crate) const fn degraded_packets_since_fallback_for_test(&self) -> u64 {
self.degraded_packets_since_fallback
}
}
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierVideoStreamDecoder<C> {
pub(crate) fn send_packet_impl(
&mut self,
packet: &VideoPacket<VideoPacketExtra, C::Buffer>,
) -> Result<(), VideoDecodeError> {
if self.scratch_pending {
return Err(VideoDecodeError::FramePending);
}
let limits = self.limits.packet_limits();
let route = match &self.state {
DecodeState::Hw(hw) if hw.records_submissions() => crate::carrier::BodyRoute::Copy,
_ => crate::carrier::BodyRoute::Submission,
};
boundary::with_ffmpeg_video_packet::<C, _>(packet, limits, route, |av_pkt| {
match &mut self.state {
DecodeState::Hw(hw) => match hw.send_packet(av_pkt) {
Ok(()) => Ok(()),
Err(Error::AllBackendsFailed(p)) => {
if p.origin().is_post_commit() {
tracing::warn!(
backend = ?p.attempts().last().map(|(b, _)| *b),
pts = ?av_pkt.pts(),
"mediadecode-ffmpeg: HW decode failed post-commit; falling back to \
software, resyncing at next keyframe — a bounded span of frames \
may be dropped at this boundary",
);
return self
.degrade_to_sw(PostCommitInput::Packet(av_pkt))
.map_err(VideoDecodeError::Decode);
}
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();
sw.send_packet(av_pkt)
.map_err(|e| VideoDecodeError::Decode(crate::decoder::software_exit(st, e)))?;
}
Ok(())
}
Err(other) => Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
let st = sw.state();
sw.send_packet(av_pkt)
.map_err(|e| VideoDecodeError::Decode(crate::decoder::software_exit(st, e)))?;
self.note_degraded_keyframe(av_pkt.is_key());
self.count_degraded_packet();
Ok(())
}
}
})
.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<(), 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 e.parks_in_decode() => 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(());
}
if self.scratch_pending {
return self.deliver_frame(dst);
}
loop {
match &mut self.state {
DecodeState::Hw(hw) => match hw.receive_frame(&mut self.hw_scratch) {
Ok(()) => {
self.scratch_pending = true;
return self.deliver_frame(dst);
}
Err(Error::AllBackendsFailed(p)) => {
if p.origin().is_post_commit() {
tracing::warn!(
backend = ?p.attempts().last().map(|(b, _)| *b),
"mediadecode-ffmpeg: HW decode failed post-commit at frame-time; \
falling back to software, resyncing at next keyframe — a bounded \
span of frames may be dropped at this boundary",
);
self
.degrade_to_sw(PostCommitInput::FrameTime)
.map_err(VideoDecodeError::Decode)?;
continue;
}
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 e.parks_in_decode() => 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(());
}
}
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 = e.parks_in_decode();
return Err(VideoDecodeError::Convert(e));
}
};
self.commit_delivery(new_frame, dst);
return Ok(());
}
Err(ffmpeg_next::Error::Eof) if self.degraded_resync_pending => {
let packets_lost = self.degraded_packets_since_fallback;
tracing::error!(
packets_lost,
"mediadecode-ffmpeg: post-commit HW->SW fallback never resynced before EOF — \
{packets_lost} packets fed to the software decoder produced no frame (no \
keyframe found across the gap); the stream tail from the fallback point was \
lost",
);
self.clear_degraded_resync();
return Err(VideoDecodeError::PostCommitNeverResynced(
PostCommitNeverResynced::new(packets_lost),
));
}
Err(e) => {
return Err(VideoDecodeError::Decode(crate::decoder::software_exit(
st, e,
)));
}
}
}
}
}
}
pub(crate) fn send_eof_impl(&mut self) -> Result<(), VideoDecodeError> {
if self.scratch_pending {
return Err(VideoDecodeError::FramePending);
}
let outcome = match &mut self.state {
DecodeState::Hw(hw) => match hw.send_eof() {
Ok(()) => Ok(()),
Err(Error::AllBackendsFailed(p)) => {
if p.origin().is_post_commit() {
tracing::warn!(
backend = ?p.attempts().last().map(|(b, _)| *b),
"mediadecode-ffmpeg: HW decode failed post-commit at EOF; falling \
back to software — a bounded span of tail frames may be dropped",
);
self
.degrade_to_sw(PostCommitInput::Eof)
.map_err(VideoDecodeError::Decode)
} else {
let rescued = p.into_unconsumed_packets();
self
.fall_back_to_sw(rescued, true)
.map_err(VideoDecodeError::Decode)
}
}
Err(other) => Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
let st = sw.state();
sw.send_eof()
.map_err(|e| VideoDecodeError::Decode(crate::decoder::software_exit(st, e)))
}
};
if outcome.is_ok() {
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;
self.clear_degraded_resync();
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 const fn is_software(&self) -> bool {
self.is_software_impl()
}
pub const fn is_hardware(&self) -> bool {
self.is_hardware_impl()
}
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<(), Self::Error> {
self.send_packet_impl(packet)
}
fn receive_frame(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, Self::Buffer>,
) -> Result<(), Self::Error> {
self.receive_frame_impl(dst)
}
fn send_eof(&mut self) -> Result<(), Self::Error> {
self.send_eof_impl()
}
fn flush(&mut self) -> Result<(), Self::Error> {
self.flush_impl()
}
}
)+ };
}
video_lane_face!(crate::View, crate::Owned);
fn open_sw_decoder(parameters: &Parameters, limits: DecoderLimits) -> Result<SwDecoder, Error> {
let (ctx, callback_state) = build_codec_context(parameters, limits)?;
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)]
#[error(
"post-commit HW->SW fallback never resynced before EOF: {packets_lost} packets fed to the \
software decoder produced no frame (no keyframe found across the gap) — the stream tail \
from the fallback point was lost"
)]
pub struct PostCommitNeverResynced {
packets_lost: u64,
}
impl PostCommitNeverResynced {
#[inline]
pub const fn new(packets_lost: u64) -> Self {
Self { packets_lost }
}
#[inline]
pub const fn packets_lost(&self) -> u64 {
self.packets_lost
}
}
#[derive(thiserror::Error, Debug, IsVariant, Unwrap, TryUnwrap)]
#[unwrap(ref, ref_mut)]
#[try_unwrap(ref, ref_mut)]
pub enum VideoDecodeError {
#[error(transparent)]
Decode(#[from] Error),
#[error(transparent)]
Convert(#[from] ConvertError),
#[error(transparent)]
PostCommitNeverResynced(#[from] PostCommitNeverResynced),
#[error(
"a decoded frame is parked and must be delivered before another packet is sent \
(call `receive_frame`, or `flush` to abandon it)"
)]
FramePending,
}
#[cfg(test)]
mod tests;