use std::collections::VecDeque;
const SW_REPLAY_FRAME_CAP: usize = 64;
use ffmpeg_next::{Packet, codec::Parameters, frame};
use mediadecode::{Timebase, decoder::VideoStreamDecoder, frame::VideoFrame, packet::VideoPacket};
use crate::{
Error, Ffmpeg, FfmpegBuffer, 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 FfmpegVideoStreamDecoder {
state: DecodeState,
parameters: Parameters,
hw_scratch: Frame,
sw_scratch: frame::Video,
sw_replay_frames: VecDeque<frame::Video>,
eof_sent: bool,
degraded_resync_pending: bool,
degraded_keyframe_seen: bool,
degraded_packets_since_fallback: u64,
time_base: Timebase,
}
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>;
}
impl HwInner for VideoDecoder {
#[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(ffmpeg_next::decoder::Video),
}
enum PostCommitInput<'a> {
Packet(&'a Packet),
FrameTime,
Eof,
}
impl FfmpegVideoStreamDecoder {
pub fn open(parameters: Parameters, time_base: Timebase) -> Result<Self, Error> {
let owned_parameters = try_clone_parameters(¶meters).map_err(Error::Ffmpeg)?;
let hw_scratch = Frame::empty()?;
let sw_scratch = alloc_av_video_frame()?;
let state =
match VideoDecoder::open(try_clone_parameters(&owned_parameters).map_err(Error::Ffmpeg)?) {
Ok(hw) => DecodeState::Hw(Box::new(hw)),
Err(Error::AllBackendsFailed(_)) => {
let sw = open_sw_decoder(&owned_parameters)?;
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,
})
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn is_software(&self) -> bool {
matches!(self.state, DecodeState::Sw(_))
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn is_hardware(&self) -> bool {
matches!(self.state, DecodeState::Hw(_))
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub fn hardware_inner(&self) -> Option<&VideoDecoder> {
match &self.state {
DecodeState::Hw(hw) => hw.as_video_decoder(),
DecodeState::Sw(_) => None,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn time_base(&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)?;
let mut local_replay: VecDeque<frame::Video> = VecDeque::new();
fn drain_into(
sw: &mut ffmpeg_next::decoder::Video,
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(Error::Ffmpeg(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, &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(Error::Ffmpeg(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, &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(Error::Ffmpeg(other)),
}
}
}
drain_into(&mut sw, &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(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)?;
let mut forwarded_keyframe = false;
let mut forwarded_packet = false;
match input {
PostCommitInput::Packet(pkt) => {
sw.send_packet(pkt).map_err(Error::Ffmpeg)?;
forwarded_keyframe = pkt.is_key();
forwarded_packet = true;
}
PostCommitInput::FrameTime => {}
PostCommitInput::Eof => {
sw.send_eof().map_err(Error::Ffmpeg)?;
}
}
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 deliver_frame(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, FfmpegBuffer>,
) -> 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 new_frame = unsafe { convert::av_frame_to_video_frame(av_frame, self.time_base) }
.map_err(VideoDecodeError::Convert)?;
*dst = new_frame;
Ok(())
}
}
#[cfg(test)]
impl FfmpegVideoStreamDecoder {
pub(crate) fn from_hw_inner_for_test(
hw: Box<dyn HwInner>,
parameters: Parameters,
time_base: Timebase,
) -> Result<Self, Error> {
let owned_parameters = try_clone_parameters(¶meters).map_err(Error::Ffmpeg)?;
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,
})
}
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 VideoStreamDecoder for FfmpegVideoStreamDecoder {
type Adapter = Ffmpeg;
type Buffer = FfmpegBuffer;
type Error = VideoDecodeError;
fn send_packet(
&mut self,
packet: &VideoPacket<VideoPacketExtra, Self::Buffer>,
) -> Result<(), Self::Error> {
let av_pkt = boundary::ffmpeg_packet_from_video_packet(packet)
.map_err(|e| VideoDecodeError::Decode(Error::Ffmpeg(e)))?;
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 {
sw.send_packet(&av_pkt)
.map_err(|e| VideoDecodeError::Decode(Error::Ffmpeg(e)))?;
}
Ok(())
}
Err(other) => Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
sw.send_packet(&av_pkt)
.map_err(|e| VideoDecodeError::Decode(Error::Ffmpeg(e)))?;
self.note_degraded_keyframe(av_pkt.is_key());
self.count_degraded_packet();
Ok(())
}
}
}
fn receive_frame(
&mut self,
dst: &mut VideoFrame<mediadecode::PixelFormat, VideoFrameExtra, Self::Buffer>,
) -> Result<(), Self::Error> {
if let Some(replayed) = self.sw_replay_frames.pop_front() {
let new_frame =
unsafe { convert::av_frame_to_video_frame(replayed.as_ptr(), self.time_base) }
.map_err(VideoDecodeError::Convert)?;
self.resync_on_frame();
*dst = new_frame;
return Ok(());
}
loop {
match &mut self.state {
DecodeState::Hw(hw) => match hw.receive_frame(&mut self.hw_scratch) {
Ok(()) => 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.pop_front() {
let new_frame =
unsafe { convert::av_frame_to_video_frame(replayed.as_ptr(), self.time_base) }
.map_err(VideoDecodeError::Convert)?;
self.resync_on_frame();
*dst = new_frame;
return Ok(());
}
}
Err(other) => return Err(VideoDecodeError::Decode(other)),
},
DecodeState::Sw(sw) => {
match sw.receive_frame(&mut self.sw_scratch) {
Ok(()) => {
let new_frame = unsafe {
convert::av_frame_to_video_frame(self.sw_scratch.as_ptr(), self.time_base)
}
.map_err(VideoDecodeError::Convert)?;
self.resync_on_frame();
*dst = new_frame;
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 { packets_lost });
}
Err(e) => return Err(VideoDecodeError::Decode(Error::Ffmpeg(e))),
}
}
}
}
}
fn send_eof(&mut self) -> Result<(), Self::Error> {
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) => sw
.send_eof()
.map_err(|e| VideoDecodeError::Decode(Error::Ffmpeg(e))),
};
if outcome.is_ok() {
self.eof_sent = true;
}
outcome
}
fn flush(&mut self) -> Result<(), Self::Error> {
self.sw_replay_frames.clear();
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(())
}
}
fn open_sw_decoder(parameters: &Parameters) -> Result<ffmpeg_next::decoder::Video, Error> {
let ctx = build_codec_context(parameters)?;
ctx.decoder().video().map_err(Error::Ffmpeg)
}
#[derive(thiserror::Error, Debug)]
pub enum VideoDecodeError {
#[error(transparent)]
Decode(#[from] Error),
#[error(transparent)]
Convert(#[from] ConvertError),
#[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"
)]
PostCommitNeverResynced {
packets_lost: u64,
},
}
#[cfg(test)]
mod tests;