ez_ffmpeg/core/context/
mod.rs

1use crate::error::AllocFrameError;
2use ffmpeg_sys_next::AVMediaType::{
3    AVMEDIA_TYPE_ATTACHMENT, AVMEDIA_TYPE_AUDIO, AVMEDIA_TYPE_DATA, AVMEDIA_TYPE_SUBTITLE,
4    AVMEDIA_TYPE_VIDEO,
5};
6// Only SideDataList::push_clone (gated to FFmpeg 8+) clones entries; the
7// free side runs on every real lane via Drop. Both symbols are FFmpeg 7.0+,
8// and docs.rs generates its bindings against an older apt FFmpeg — so like
9// every other >=7.0 symbol in this crate they stay out of docsrs builds.
10#[cfg(ffmpeg_8_0)]
11use ffmpeg_sys_next::av_frame_side_data_clone;
12#[cfg(not(docsrs))]
13use ffmpeg_sys_next::av_frame_side_data_free;
14use ffmpeg_sys_next::{
15    av_freep, av_gettime_relative, avcodec_free_context, avformat_close_input,
16    avformat_free_context, avio_closep, avio_context_free, AVCodecContext, AVFormatContext,
17    AVFrameSideData, AVIOContext, AVMediaType, AVRational, AVStream, AVFMT_NOFILE,
18};
19use std::ffi::c_void;
20use std::ptr::null_mut;
21use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
22use std::sync::Arc;
23
24/// How long output I/O may continue after STATUS_END before the interrupt
25/// callback cuts it: long enough for a healthy muxer to finish its trailer,
26/// short enough that stop() on a dead network peer returns within a second.
27/// While a muxer holds an [`OutputFinalizeGuard`] the window does not apply
28/// at all — a `movflags=+faststart` trailer rewrites the whole file and can
29/// legitimately take far longer than any fixed grace.
30const OUTPUT_END_GRACE_US: i64 = 500_000;
31
32/// Default size of the AVIO buffer backing a custom read/write callback. FFmpeg
33/// hands the callback one buffer-sized chunk at a time, so a larger buffer means
34/// fewer Rust↔FFmpeg round-trips for sequential/network IO; the 64 KiB default
35/// keeps first-packet latency low for live/low-latency use. Configurable per
36/// Input/Output via set_io_buffer_size (sys-06).
37pub(crate) const DEFAULT_CUSTOM_IO_BUFFER_SIZE: usize = 64 * 1024;
38
39/// Shared state behind the AVIO interrupt callbacks (fftools installs
40/// decode_interrupt_cb on inputs and outputs, ffmpeg_mux_init.c:3326,3371).
41///
42/// Inputs are interrupted as soon as the scheduler is stopping: nothing
43/// meaningful is read after that. Outputs distinguish the terminal states:
44/// STATUS_ABORT cuts I/O immediately (the caller gave up on the files), while
45/// STATUS_END grants a grace window so trailers still get written — only an
46/// output that stays blocked past the window (dead network peer) is cut.
47pub(crate) struct InterruptState {
48    scheduler_status: Arc<AtomicUsize>,
49    // Microsecond timestamp of the first output callback that observed
50    // STATUS_END; 0 = not observed yet.
51    end_grace_start_us: AtomicI64,
52    /// Number of mux workers currently inside their finalize window
53    /// (av_write_trailer through the output-context free). While > 0,
54    /// STATUS_END does not interrupt output I/O — only STATUS_ABORT does: a
55    /// graceful stop() must not corrupt a trailer that is being written
56    /// (H3), and abort() stays the hard-cancel escape hatch.
57    finalizing_outputs: AtomicUsize,
58}
59
60impl InterruptState {
61    pub(crate) fn new(scheduler_status: Arc<AtomicUsize>) -> Self {
62        Self {
63            scheduler_status,
64            end_grace_start_us: AtomicI64::new(0),
65            finalizing_outputs: AtomicUsize::new(0),
66        }
67    }
68
69    /// Marks one output as finalizing for the guard's lifetime.
70    pub(crate) fn begin_output_finalize(self: &Arc<Self>) -> OutputFinalizeGuard {
71        self.finalizing_outputs.fetch_add(1, Ordering::AcqRel);
72        OutputFinalizeGuard {
73            state: Arc::clone(self),
74        }
75    }
76
77    fn should_interrupt_input(&self) -> bool {
78        crate::core::scheduler::ffmpeg_scheduler::is_stopping(
79            self.scheduler_status.load(Ordering::Acquire),
80        )
81    }
82
83    fn should_interrupt_output(&self) -> bool {
84        let status = self.scheduler_status.load(Ordering::Acquire);
85        if status == crate::core::scheduler::ffmpeg_scheduler::STATUS_ABORT {
86            return true;
87        }
88        if status != crate::core::scheduler::ffmpeg_scheduler::STATUS_END {
89            return false;
90        }
91        if self.finalizing_outputs.load(Ordering::Acquire) > 0 {
92            // A trailer (e.g. a faststart moov rewrite) is in flight somewhere
93            // on this scheduler; only abort() may cut output I/O now. Checked
94            // BEFORE the grace CAS so the clock does not even start while a
95            // finalize is running; once the last guard drops, a still-blocked
96            // OTHER output starts its grace normally.
97            return false;
98        }
99        let now = unsafe { av_gettime_relative() };
100        let start = match self.end_grace_start_us.compare_exchange(
101            0,
102            now,
103            Ordering::AcqRel,
104            Ordering::Acquire,
105        ) {
106            Ok(_) => now,
107            Err(previous) => previous,
108        };
109        now - start > OUTPUT_END_GRACE_US
110    }
111}
112
113/// RAII finalize marker. The decrement is mandatory even on unwind — a stuck
114/// counter would exempt every later STATUS_END grace cut for this scheduler,
115/// and stop() on a blocked sibling output would hang forever.
116pub(crate) struct OutputFinalizeGuard {
117    state: Arc<InterruptState>,
118}
119
120impl Drop for OutputFinalizeGuard {
121    fn drop(&mut self) {
122        self.state.finalizing_outputs.fetch_sub(1, Ordering::AcqRel);
123    }
124}
125
126/// # Safety
127/// `opaque` must point to an `InterruptState` that outlives every
128/// AVFormatContext carrying this callback (the owning FfmpegContext holds the
129/// Arc and outlives all worker threads).
130pub(crate) unsafe extern "C" fn input_interrupt_cb(opaque: *mut c_void) -> libc::c_int {
131    let state = &*(opaque as *const InterruptState);
132    state.should_interrupt_input() as libc::c_int
133}
134
135/// # Safety
136/// Same contract as [`input_interrupt_cb`].
137pub(crate) unsafe extern "C" fn output_interrupt_cb(opaque: *mut c_void) -> libc::c_int {
138    let state = &*(opaque as *const InterruptState);
139    let interrupt = state.should_interrupt_output();
140    // This callback runs on the muxer's own thread, inside the retry loop
141    // of whatever blocking I/O it is cutting, so an election here is the
142    // one place that can attribute the cut to the exact write in flight.
143    #[cfg(test)]
144    if interrupt {
145        crate::core::scheduler::tcp_write_probe::note_cut_election();
146    }
147    interrupt as libc::c_int
148}
149
150/// Gate for the muxer's deferred start. fftools: `SchMux.mux_started` +
151/// `PreMuxQueue` under `Scheduler.mux_ready_lock` (ffmpeg_sched.c).
152///
153/// Until the muxer thread is running, encoders park packets in a bounded
154/// pre-queue; at start the muxer drains that queue and flips `started`.
155/// Without a lock the flip races the send: an encoder that read
156/// `started == false` can enqueue into the pre-queue AFTER the drain
157/// finished, and that packet is never delivered. Pre-queue sends therefore
158/// happen under the same lock as the drain-and-flip.
159pub(crate) struct MuxStartGate {
160    started: std::sync::atomic::AtomicBool,
161    lock: std::sync::Mutex<()>,
162}
163
164/// What a gated pre-queue send resolved to.
165pub(crate) enum PreSendOutcome {
166    /// Parked in the pre-queue; the drain will deliver it.
167    Sent,
168    /// The gate opened first: send to the live queue instead.
169    Started(PacketBox),
170    /// Pre-queue full: park on the queue condvar and retry (the gate lock
171    /// must not be held across the wait, or the drain could never run).
172    Full(PacketBox),
173    /// Pre-queue receiver is gone (muxer never started).
174    Disconnected(PacketBox),
175}
176
177impl MuxStartGate {
178    pub(crate) fn new() -> Self {
179        Self {
180            started: std::sync::atomic::AtomicBool::new(false),
181            lock: std::sync::Mutex::new(()),
182        }
183    }
184
185    pub(crate) fn is_started(&self) -> bool {
186        self.started.load(Ordering::Acquire)
187    }
188
189    /// Runs the pre-queue drain and opens the gate as one atomic step.
190    pub(crate) fn start_with(&self, drain: impl FnOnce()) {
191        let _guard = self.lock.lock().unwrap();
192        drain();
193        self.started.store(true, Ordering::Release);
194    }
195
196    /// Attempts a pre-queue send while the gate is verifiably closed.
197    pub(crate) fn send_pre(
198        &self,
199        pre_sender: &pre_mux_queue::PreMuxQueueSender,
200        packet_box: PacketBox,
201    ) -> PreSendOutcome {
202        let _guard = self.lock.lock().unwrap();
203        if self.started.load(Ordering::Acquire) {
204            return PreSendOutcome::Started(packet_box);
205        }
206        match pre_sender.try_push(packet_box) {
207            pre_mux_queue::PreQueueTryPush::Sent => PreSendOutcome::Sent,
208            pre_mux_queue::PreQueueTryPush::Full(pb) => PreSendOutcome::Full(pb),
209            pre_mux_queue::PreQueueTryPush::Disconnected(pb) => PreSendOutcome::Disconnected(pb),
210        }
211    }
212}
213
214use ffmpeg_context::{InputOpaque, OutputOpaque};
215
216/// The **ffmpeg_context** module is responsible for assembling FFmpeg’s configuration:
217/// inputs, outputs, codecs, filters, and other parameters needed to construct a
218/// complete media processing pipeline.
219///
220/// # Example
221/// ```rust,ignore
222///
223/// // Build an FFmpeg context with one input, some filter settings, and one output
224/// let context = FfmpegContext::builder()
225///     .input("test.mp4")
226///     .filter_desc("hue=s=0")
227///     .output("output.mp4")
228///     .build()
229///     .unwrap();
230/// // The context now holds all info needed for an FFmpeg job.
231/// ```
232pub mod ffmpeg_context;
233
234/// The **ffmpeg_context_builder** module defines the builder pattern for creating
235/// [`FfmpegContext`](ffmpeg_context::FfmpegContext) objects.
236///
237/// It exposes the [`FfmpegContextBuilder`](ffmpeg_context_builder::FfmpegContextBuilder) struct, which allows you to:
238/// - Configure multiple [`Input`](input::Input) and
239///   [`Output`](output::Output) streams.
240/// - Attach filter descriptions via [`FilterComplex`](crate::core::context::filter_complex::FilterComplex)
241///   or inline strings (e.g., `"scale=1280:720"`, `"hue=s=0"`).
242/// - Produce a finished `FfmpegContext` that can then be executed by
243///   [`FfmpegScheduler`](crate::core::scheduler::ffmpeg_scheduler::FfmpegScheduler).
244///
245/// # Examples
246///
247/// ```rust,ignore
248/// // 1. Create a builder (usually via FfmpegContext::builder())
249/// let builder = FfmpegContext::builder();
250///
251/// // 2. Add inputs, outputs, and filters
252/// let ffmpeg_context = builder
253///     .input("input.mp4")
254///     .filter_desc("hue=s=0")
255///     .output("output.mp4")
256///     .build()
257///     .expect("Failed to build FfmpegContext");
258///
259/// // 3. Use `ffmpeg_context` with FfmpegScheduler (e.g., `.start()` and `.wait()`).
260/// ```
261pub mod ffmpeg_context_builder;
262
263/// The **input** module defines the [`Input`](crate::core::context::input::Input) struct,
264/// representing an FFmpeg input source. An input can be:
265/// - A file path or URL (e.g., `"video.mp4"`, `rtmp://example.com/live/stream`).
266/// - A **custom data source** via a `read_callback` (and optionally `seek_callback`) for
267///   advanced scenarios like in-memory buffers or network protocols.
268///
269/// You can also specify **frame pipelines** to apply custom [`FrameFilter`](crate::core::filter::frame_filter::FrameFilter)
270/// transformations **after decoding** but **before** the frames move on to the rest of the pipeline.
271///
272/// # Example
273///
274/// ```rust,ignore
275/// use ez_ffmpeg::core::context::input::Input;
276///
277/// // Basic file or network URL:
278/// let file_input: Input = "example.mp4".into();
279///
280/// // Or a custom read callback:
281/// let custom_input = Input::new_by_read_callback(|buf| {
282///     // Fill `buf` with data from your source
283///     // Return the number of bytes read, or negative for errors
284///     0
285/// });
286/// ```
287pub mod input;
288
289/// The **output** module defines the [`Output`](crate::core::context::output::Output) struct,
290/// representing an FFmpeg output destination. An output may be:
291/// - A file path or URL (e.g., `"output.mp4"`, `rtmp://...`).
292/// - A **custom write callback** that processes encoded data (e.g., storing it
293///   in-memory or sending it over a custom network protocol).
294///
295/// You can specify additional details such as:
296/// - **Container format** (e.g., `"mp4"`, `"flv"`, `"mkv"`).
297/// - **Video/Audio/Subtitle codecs** (e.g., `"h264"`, `"aac"`, `"mov_text"`).
298/// - **Frame pipelines** to apply [`FrameFilter`](crate::core::filter::frame_filter::FrameFilter)
299///   transformations **before encoding**.
300///
301/// # Example
302///
303/// ```rust,ignore
304/// use ez_ffmpeg::core::context::output::Output;
305///
306/// // Basic file/URL output:
307/// let file_output: Output = "output.mp4".into();
308///
309/// // Or a custom write callback:
310/// let custom_output = Output::new_by_write_callback(|encoded_data| {
311///     // Write `encoded_data` somewhere
312///     encoded_data.len() as i32
313/// }).set_format("mp4");
314/// ```
315pub mod output;
316
317/// The **filter_complex** module defines the [`FilterComplex`](crate::core::context::filter_complex::FilterComplex)
318/// struct, which encapsulates one or more FFmpeg filter descriptions (e.g., `"scale=1280:720"`,
319/// `"hue=s=0"`, etc.). You can use `FilterComplex` to construct more advanced or multi-step
320/// filter graphs than simple inline strings allow.
321///
322/// `FilterComplex` can also associate a particular hardware device (e.g., for GPU-based
323/// filtering) via `hw_device`.
324///
325/// # Example
326///
327/// ```rust,ignore
328/// use ez_ffmpeg::core::context::filter_complex::FilterComplex;
329///
330/// // Build a FilterComplex from a string:
331/// let my_filters = FilterComplex::from("scale=1280:720");
332///
333/// // Optionally specify a hardware device (e.g., "cuda"):
334/// // my_filters.set_hw_device("cuda");
335/// ```
336pub mod filter_complex;
337
338pub(super) mod attachment;
339pub(super) mod decoder_stream;
340pub(super) mod demuxer;
341pub(super) mod encoder_stream;
342pub(super) mod filter_graph;
343pub(super) mod frame_source;
344pub(super) mod input_filter;
345pub(super) mod muxer;
346pub(super) mod obj_pool;
347pub(super) mod output_filter;
348pub(super) mod pre_mux_queue;
349
350/// The **null_output** module provides a custom null output implementation for FFmpeg
351/// that discards all data while supporting seeking.
352///
353/// It exposes the [`create_null_output`](null_output::create_null_output) function, which returns an
354/// [`Output`](crate::Output) object configured to:
355/// - Discard all written data, behaving like `/dev/null`.
356/// - Maintain a seekable position state using atomic operations for thread-safe, high-performance access.
357/// - Support scenarios such as testing or processing streaming inputs (e.g., RTMP) where no output file is needed.
358///
359/// # Usage Scenario
360/// This module is useful when processing FFmpeg input streams without generating an output file, such as
361/// when handling RTMP streams that require a seekable output format like MP4, even if the output is discarded.
362///
363/// # Examples
364///
365/// ```rust,ignore
366/// use ez_ffmpeg::Output;
367/// let output: Output = create_null_output();
368/// // Pass `output` to an FFmpeg context for processing
369/// ```
370///
371/// # Performance
372/// - Utilizes `AtomicU64` with `Relaxed` ordering for lock-free position tracking, ensuring efficient concurrent access.
373/// - Write and seek operations are optimized to minimize overhead by avoiding locks.
374///
375/// # Notes
376/// - The default output format is "mp4", but this can be modified using `set_format` as needed.
377/// - Write operations assume individual buffers do not exceed `i32::MAX` bytes, which aligns with typical FFmpeg usage.
378pub mod null_output;
379
380pub(crate) struct CodecContext {
381    inner: *mut AVCodecContext,
382}
383
384// SAFETY: CodecContext can be sent to another thread. It is a single-owner
385// handle: application-side lifecycle API calls (open/decode/flush/free) go
386// through its one owner (a worker's resource struct or an open-function
387// local on failure paths), never through aliases on other Rust threads.
388// Rust code DOES also run on FFmpeg-managed threads — decode callbacks like
389// `get_format` receive a context pointer (often a per-thread copy) there —
390// but libavcodec synchronizes those accesses, and `avcodec_free_context`
391// (Drop) quiesces all callbacks and joins FFmpeg's workers before returning.
392unsafe impl Send for CodecContext {}
393
394impl CodecContext {
395    pub(crate) fn new(avcodec_context: *mut AVCodecContext) -> Self {
396        Self {
397            inner: avcodec_context,
398        }
399    }
400
401    pub(crate) fn as_mut_ptr(&self) -> *mut AVCodecContext {
402        self.inner
403    }
404
405    pub(crate) fn as_ptr(&self) -> *const AVCodecContext {
406        self.inner as *const AVCodecContext
407    }
408}
409
410impl Drop for CodecContext {
411    fn drop(&mut self) {
412        unsafe {
413            avcodec_free_context(&mut self.inner);
414        }
415    }
416}
417
418#[derive(Copy, Clone)]
419pub(crate) struct Stream {
420    pub(crate) inner: *mut AVStream,
421}
422
423// SAFETY: Stream can be sent to another thread. The raw AVStream pointer is owned
424// by the parent AVFormatContext, and the crate ensures the format context outlives
425// all Stream references.
426unsafe impl Send for Stream {}
427
428pub(crate) struct FrameBox {
429    pub(crate) frame: ffmpeg_next::Frame,
430    // stream copy or filtergraph
431    pub(crate) frame_data: FrameData,
432}
433
434// SAFETY: FrameBox can be sent to another thread. It contains an ffmpeg_next::Frame
435// (which wraps AVFrame) and FrameData, both of which are only accessed from the owning thread.
436unsafe impl Send for FrameBox {}
437
438pub fn frame_alloc() -> crate::error::Result<ffmpeg_next::Frame> {
439    unsafe {
440        let frame = ffmpeg_next::Frame::empty();
441        if frame.as_ptr().is_null() {
442            return Err(AllocFrameError::OutOfMemory.into());
443        }
444        Ok(frame)
445    }
446}
447
448pub fn null_frame() -> ffmpeg_next::Frame {
449    unsafe { ffmpeg_next::Frame::wrap(null_mut()) }
450}
451
452/// Owned array of global (stream-level) side data entries — the Rust shape of
453/// the `AVFrameSideData **side_data / int nb_side_data` pairs that fftools
454/// threads through InputFilterPriv, OutputFilterPriv and FrameData
455/// (ffmpeg_filter.c, commits e61b9d4094 / 7b18beb477). Entries are
456/// deep-cloned in and freed exactly once on drop.
457pub(crate) struct SideDataList {
458    entries: *mut *mut AVFrameSideData,
459    count: i32,
460}
461
462impl SideDataList {
463    pub(crate) fn new() -> Self {
464        Self {
465            entries: null_mut(),
466            count: 0,
467        }
468    }
469
470    /// Deep-clones one entry onto the list (flags as in
471    /// `av_frame_side_data_clone`). Returns 0 or a negative AVERROR.
472    #[cfg(ffmpeg_8_0)]
473    pub(crate) fn push_clone(&mut self, sd: *const AVFrameSideData, flags: u32) -> i32 {
474        unsafe { av_frame_side_data_clone(&mut self.entries, &mut self.count, sd, flags) }
475    }
476
477    pub(crate) fn clear(&mut self) {
478        // docs.rs bindings predate av_frame_side_data_free (FFmpeg 7.0), but
479        // there the list is provably empty — push_clone is ffmpeg_8_0-gated
480        // and that cfg never exists on docs.rs — so skipping the call is the
481        // exact semantics, not a stub.
482        #[cfg(not(docsrs))]
483        unsafe {
484            av_frame_side_data_free(&mut self.entries, &mut self.count)
485        }
486    }
487
488    #[cfg(ffmpeg_8_0)]
489    pub(crate) fn len(&self) -> i32 {
490        self.count
491    }
492
493    /// Raw entry array for FFmpeg parameter structs
494    /// (e.g. `AVBufferSrcParameters.side_data`); callees deep-copy what they
495    /// need, the list keeps ownership. Takes `&mut self` so handing out a
496    /// `*mut` view is justified by the signature, not by convention — which
497    /// also keeps the Sync claim honest (shared refs never leak mutability).
498    #[cfg(ffmpeg_8_0)]
499    pub(crate) fn as_mut_ptr(&mut self) -> *mut *mut AVFrameSideData {
500        self.entries
501    }
502
503    #[cfg(ffmpeg_8_0)]
504    pub(crate) fn iter(&self) -> impl Iterator<Item = *const AVFrameSideData> + '_ {
505        (0..self.count)
506            .map(|i| unsafe { *self.entries.offset(i as isize) as *const AVFrameSideData })
507    }
508}
509
510impl Drop for SideDataList {
511    fn drop(&mut self) {
512        self.clear();
513    }
514}
515
516// SAFETY: the list exclusively owns its deep-cloned entries. All mutation
517// happens on the thread that is building it; once frozen behind an Arc it is
518// only read, and readers clone entries out instead of mutating — so both
519// moving it across threads and sharing references are sound.
520unsafe impl Send for SideDataList {}
521unsafe impl Sync for SideDataList {}
522
523#[derive(Clone)]
524pub(crate) struct FrameData {
525    pub(crate) framerate: Option<AVRational>,
526    pub(crate) bits_per_raw_sample: i32,
527    pub(crate) input_stream_width: i32,
528    pub(crate) input_stream_height: i32,
529    /// Owned copy of the decoder's subtitle header (e.g. ASS script info),
530    /// shared across fan-out sends without reallocation. Owning the bytes
531    /// keeps the header valid after the decoder context is freed.
532    pub(crate) subtitle_header: Option<Arc<[u8]>>,
533
534    pub(crate) fg_input_index: usize,
535
536    /// Graph-level global side data snapshot, attached by the filtergraph
537    /// output for the frame that opens the encoder and consumed by
538    /// `enc_open` on FFmpeg 8+ (fftools FrameData.side_data, 7b18beb477).
539    /// Always `None` on FFmpeg 7.x, where the encoder scans frame side data
540    /// instead (n7.1 behavior) — hence dead there by design.
541    #[cfg_attr(not(ffmpeg_8_0), allow(dead_code))]
542    pub(crate) side_data: Option<Arc<SideDataList>>,
543}
544// Send + Sync are auto-derived. For side_data that rests on SideDataList's
545// hand-written unsafe impls (frozen-behind-Arc discipline); every other
546// field is plain owned data.
547
548pub(crate) struct PacketBox {
549    pub(crate) packet: ffmpeg_next::Packet,
550    pub(crate) packet_data: PacketData,
551}
552
553// SAFETY: PacketBox can be sent to another thread. It contains an ffmpeg_next::Packet
554// and PacketData, both only accessed from the owning thread.
555unsafe impl Send for PacketBox {}
556
557// optionally attached as opaque_ref to decoded AVFrames
558#[derive(Clone, Copy)]
559pub(crate) struct PacketData {
560    // demuxer-estimated dts in AV_TIME_BASE_Q,
561    // to be used when real dts is missing
562    pub(crate) dts_est: i64,
563    pub(crate) codec_type: AVMediaType,
564    pub(crate) output_stream_index: i32,
565    pub(crate) is_copy: bool,
566}
567
568pub(crate) fn out_fmt_ctx_free(out_fmt_ctx: *mut AVFormatContext, is_set_write_callback: bool) {
569    if out_fmt_ctx.is_null() {
570        return;
571    }
572    unsafe {
573        let custom_io_teardown_panic = if is_set_write_callback {
574            // Detach pb before reclaiming it, then retain a callback-state
575            // teardown panic until the owning format context is also freed.
576            // free_output_opaque suppresses its stable panic when this function
577            // is already running during an unwind.
578            let avio_ctx = (*out_fmt_ctx).pb;
579            (*out_fmt_ctx).pb = null_mut();
580            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
581                free_output_opaque(avio_ctx)
582            }))
583            .err()
584        } else if (*(*out_fmt_ctx).oformat).flags & AVFMT_NOFILE == 0 {
585            // AVFMT_NOFILE lives on the *output format* flags, not the context flags
586            // (where the same bit 0x1 is AVFMT_FLAG_GENPTS). Reading it off the
587            // context could skip avio_closep and leak the pb when GENPTS is set.
588            let mut pb = (*out_fmt_ctx).pb;
589            if !pb.is_null() {
590                avio_closep(&mut pb);
591            }
592            None
593        } else {
594            None
595        };
596        avformat_free_context(out_fmt_ctx);
597        if let Some(payload) = custom_io_teardown_panic {
598            std::panic::resume_unwind(payload);
599        }
600    }
601}
602
603/// Reports whether a custom input AVIO callback panicked and poisoned its state.
604///
605/// A null AVIO context or null opaque pointer reports `false`.
606///
607/// # Safety
608/// A non-null `avio_ctx` must be a live custom-input `AVIOContext` created by
609/// this crate, and its opaque pointer must still belong to that context. The
610/// caller must have exclusive access while reading the non-atomic poison flag.
611pub(crate) unsafe fn input_custom_io_is_poisoned(avio_ctx: *mut AVIOContext) -> bool {
612    if avio_ctx.is_null() {
613        return false;
614    }
615    let opaque_ptr = (*avio_ctx).opaque as *const InputOpaque;
616    !opaque_ptr.is_null() && (*opaque_ptr).poisoned
617}
618
619/// Reports whether a custom output AVIO callback panicked and poisoned its state.
620///
621/// A null AVIO context or null opaque pointer reports `false`.
622///
623/// # Safety
624/// A non-null `avio_ctx` must be a live custom-output `AVIOContext` created by
625/// this crate, and its opaque pointer must still belong to that context. The
626/// caller must have exclusive access while reading the non-atomic poison flag.
627pub(crate) unsafe fn output_custom_io_is_poisoned(avio_ctx: *mut AVIOContext) -> bool {
628    if avio_ctx.is_null() {
629        return false;
630    }
631    let opaque_ptr = (*avio_ctx).opaque as *const OutputOpaque;
632    !opaque_ptr.is_null() && (*opaque_ptr).poisoned
633}
634
635/// Drops one erased callback box without allowing its destructor panic to skip
636/// sibling callback disposal or AVIO reclamation.
637fn drop_custom_io_callback_contained<T>(callback: T) -> bool {
638    match std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || drop(callback))) {
639        Ok(()) => false,
640        Err(payload) => {
641            crate::core::packet_sink::dispose_panic_payload(payload);
642            true
643        }
644    }
645}
646
647unsafe fn dispose_output_opaque_callbacks(opaque_ptr: *mut OutputOpaque) -> bool {
648    if opaque_ptr.is_null() {
649        return false;
650    }
651    let OutputOpaque {
652        write,
653        seek,
654        poisoned: _,
655    } = *Box::from_raw(opaque_ptr);
656    let mut panicked = drop_custom_io_callback_contained(write);
657    if let Some(seek) = seek {
658        panicked |= drop_custom_io_callback_contained(seek);
659    }
660    panicked
661}
662
663unsafe fn dispose_input_opaque_callbacks(opaque_ptr: *mut InputOpaque) -> bool {
664    if opaque_ptr.is_null() {
665        return false;
666    }
667    let InputOpaque {
668        read,
669        seek,
670        poisoned: _,
671    } = *Box::from_raw(opaque_ptr);
672    let mut panicked = drop_custom_io_callback_contained(read);
673    if let Some(seek) = seek {
674        panicked |= drop_custom_io_callback_contained(seek);
675    }
676    panicked
677}
678
679pub(crate) unsafe fn free_output_opaque(mut avio_ctx: *mut AVIOContext) {
680    if avio_ctx.is_null() {
681        return;
682    }
683
684    let already_panicking = std::thread::panicking();
685    let mut callback_state_panicked = false;
686    let opaque_ptr = (*avio_ctx).opaque as *mut OutputOpaque;
687    (*avio_ctx).opaque = null_mut();
688    callback_state_panicked |= dispose_output_opaque_callbacks(opaque_ptr);
689
690    if !(*avio_ctx).buffer.is_null() {
691        av_freep(&mut (*avio_ctx).buffer as *mut _ as *mut c_void);
692    }
693    avio_context_free(&mut avio_ctx);
694
695    // A destructor panic on a healthy path is a worker failure, but every
696    // callback box and native allocation must be reclaimed before publishing
697    // it. During an existing unwind the original panic remains authoritative.
698    if callback_state_panicked && !already_panicking {
699        panic!("custom-IO output callback state panicked during teardown");
700    }
701}
702
703pub(crate) fn in_fmt_ctx_free(mut in_fmt_ctx: *mut AVFormatContext, is_set_read_callback: bool) {
704    if in_fmt_ctx.is_null() {
705        return;
706    }
707    unsafe {
708        // Close the input FIRST: the demuxer's read_close may still touch
709        // s->pb (the official custom-IO example frees the AVIOContext only
710        // after avformat_close_input). With AVFMT_FLAG_CUSTOM_IO the close
711        // leaves pb alone, so capture it beforehand and free it after.
712        let avio_ctx = if is_set_read_callback {
713            (*in_fmt_ctx).pb
714        } else {
715            null_mut()
716        };
717        avformat_close_input(&mut in_fmt_ctx);
718        free_input_opaque(avio_ctx);
719    }
720}
721
722pub(crate) unsafe fn free_input_opaque(mut avio_ctx: *mut AVIOContext) {
723    if avio_ctx.is_null() {
724        return;
725    }
726
727    let already_panicking = std::thread::panicking();
728    let mut callback_state_panicked = false;
729    let opaque_ptr = (*avio_ctx).opaque as *mut InputOpaque;
730    (*avio_ctx).opaque = null_mut();
731    callback_state_panicked |= dispose_input_opaque_callbacks(opaque_ptr);
732
733    if !(*avio_ctx).buffer.is_null() {
734        av_freep(&mut (*avio_ctx).buffer as *mut _ as *mut c_void);
735    }
736    avio_context_free(&mut avio_ctx);
737
738    if callback_state_panicked && !already_panicking {
739        panic!("custom-IO input callback state panicked during teardown");
740    }
741}
742
743/// RAII guard for a partially-initialized `AVFormatContext` (input or output).
744///
745/// During `open_input_file` / `open_output_file` the raw context (and, for
746/// custom-IO, its `AVIOContext` + callback `Box`) is owned by nobody until a
747/// [`Demuxer`]/[`Muxer`] takes it. Any `?`/early return in that window used to
748/// leak it. [`arm`](FmtCtxGuard::arm) it once the context is valid; on drop it
749/// frees via the same [`in_fmt_ctx_free`]/[`out_fmt_ctx_free`] paths a
750/// success-path [`crate::raw::FormatContext`] drop uses — unless
751/// [`release`](FmtCtxGuard::release) is called when ownership transfers.
752///
753/// The teardown path is selected by [`crate::raw::Mode`] (the same discriminant
754/// `FormatContext` carries), replacing the two former bool-keyed guards
755/// (`OutFmtCtxGuard`/`InFmtCtxGuard`).
756///
757/// Unlike `FormatContext`, this guard is re-armable and covers contexts that are
758/// only *partially* initialized (allocated but not yet opened, or mid custom-IO
759/// setup) — which is why it stays a separate type rather than reusing
760/// `FormatContext`'s already-opened constructors.
761pub(crate) struct FmtCtxGuard {
762    ctx: *mut AVFormatContext,
763    mode: crate::raw::Mode,
764}
765
766impl FmtCtxGuard {
767    pub(crate) fn disarmed() -> Self {
768        // `mode` is irrelevant while `ctx` is null (Drop no-ops on null).
769        Self {
770            ctx: null_mut(),
771            mode: crate::raw::Mode::Input,
772        }
773    }
774
775    /// Take ownership of a now-valid context so any early return frees it, with
776    /// the teardown path selected by `mode`.
777    pub(crate) fn arm(&mut self, ctx: *mut AVFormatContext, mode: crate::raw::Mode) {
778        self.ctx = ctx;
779        self.mode = mode;
780    }
781
782    /// Relinquish ownership (a Demuxer/Muxer/FormatContext now owns the context).
783    pub(crate) fn release(&mut self) -> *mut AVFormatContext {
784        let ctx = self.ctx;
785        self.ctx = null_mut();
786        ctx
787    }
788
789    /// Relinquish ownership into a [`crate::raw::FormatContext`] carrying the
790    /// same teardown [`crate::raw::Mode`] this guard was armed with: the mode
791    /// is decided once, at the arm site, and preserved end-to-end — callers
792    /// never re-infer it from output shape.
793    ///
794    /// Panics if the guard is disarmed (nothing to release); callers invoke
795    /// this only after a successful `arm`.
796    pub(crate) fn release_into(&mut self) -> crate::raw::FormatContext {
797        let mode = self.mode;
798        let ctx = self.release();
799        assert!(!ctx.is_null(), "release_into on a disarmed FmtCtxGuard");
800        // SAFETY: `ctx` was armed as a valid, fully-initialized context of
801        // exactly this mode (for OutputCustomIo the AVIO/pb wiring precedes
802        // the arm); ownership transfers to the returned FormatContext and the
803        // guard is now disarmed, so no double free is possible.
804        unsafe { crate::raw::FormatContext::from_mode(ctx, mode) }
805    }
806}
807
808impl Drop for FmtCtxGuard {
809    fn drop(&mut self) {
810        if self.ctx.is_null() {
811            return;
812        }
813        // Same dispatch as `FormatContext::Drop`.
814        match self.mode {
815            crate::raw::Mode::Input => in_fmt_ctx_free(self.ctx, false),
816            crate::raw::Mode::InputCustomIo => in_fmt_ctx_free(self.ctx, true),
817            crate::raw::Mode::Output => out_fmt_ctx_free(self.ctx, false),
818            crate::raw::Mode::OutputCustomIo => out_fmt_ctx_free(self.ctx, true),
819        }
820    }
821}
822
823#[allow(dead_code)]
824pub(crate) fn type_to_linklabel(media_type: AVMediaType, index: usize) -> Option<String> {
825    match media_type {
826        AVMediaType::AVMEDIA_TYPE_UNKNOWN => None,
827        AVMEDIA_TYPE_VIDEO => Some(format!("{index}:v")),
828        AVMEDIA_TYPE_AUDIO => Some(format!("{index}:a")),
829        AVMEDIA_TYPE_DATA => Some(format!("{index}:d")),
830        AVMEDIA_TYPE_SUBTITLE => Some(format!("{index}:s")),
831        AVMEDIA_TYPE_ATTACHMENT => Some(format!("{index}:t")),
832        AVMediaType::AVMEDIA_TYPE_NB => None,
833    }
834}
835
836#[cfg(test)]
837mod tests {
838    use super::*;
839    use crate::core::scheduler::ffmpeg_scheduler::{STATUS_ABORT, STATUS_END, STATUS_RUN};
840
841    unsafe fn test_avio(opaque: *mut c_void, write_flag: libc::c_int) -> *mut AVIOContext {
842        const BUFFER_SIZE: usize = 128;
843        let buffer = ffmpeg_sys_next::av_malloc(BUFFER_SIZE) as *mut u8;
844        assert!(!buffer.is_null(), "test AVIO buffer allocation failed");
845        let avio_ctx = ffmpeg_sys_next::avio_alloc_context(
846            buffer,
847            BUFFER_SIZE as libc::c_int,
848            write_flag,
849            opaque,
850            None,
851            None,
852            None,
853        );
854        if avio_ctx.is_null() {
855            ffmpeg_sys_next::av_free(buffer.cast());
856            panic!("test AVIO context allocation failed");
857        }
858        avio_ctx
859    }
860
861    fn panic_message(payload: &(dyn std::any::Any + Send)) -> &str {
862        payload
863            .downcast_ref::<&str>()
864            .copied()
865            .or_else(|| payload.downcast_ref::<String>().map(String::as_str))
866            .unwrap_or("non-string panic payload")
867    }
868
869    struct CallbackDropBomb(Arc<AtomicUsize>);
870
871    impl Drop for CallbackDropBomb {
872        fn drop(&mut self) {
873            self.0.fetch_add(1, Ordering::SeqCst);
874            panic!("test callback capture destructor");
875        }
876    }
877
878    struct FreeInputAvioOnDrop(*mut AVIOContext);
879
880    impl Drop for FreeInputAvioOnDrop {
881        fn drop(&mut self) {
882            unsafe { free_input_opaque(self.0) };
883        }
884    }
885
886    struct FreeOutputAvioOnDrop(*mut AVIOContext);
887
888    impl Drop for FreeOutputAvioOnDrop {
889        fn drop(&mut self) {
890            unsafe { free_output_opaque(self.0) };
891        }
892    }
893
894    fn state_with(status: usize) -> Arc<InterruptState> {
895        Arc::new(InterruptState::new(Arc::new(AtomicUsize::new(status))))
896    }
897
898    /// Rewinds the grace clock so the window is already expired — no sleeps.
899    fn expire_grace(state: &InterruptState) {
900        let past = unsafe { av_gettime_relative() } - OUTPUT_END_GRACE_US - 1;
901        state.end_grace_start_us.store(past, Ordering::Release);
902    }
903
904    #[test]
905    fn output_grace_cut_fires_after_the_window() {
906        let state = state_with(STATUS_END);
907        expire_grace(&state);
908        assert!(state.should_interrupt_output());
909    }
910
911    #[test]
912    fn finalize_guard_holds_the_grace_cut_open() {
913        let state = state_with(STATUS_END);
914        expire_grace(&state);
915        let guard = state.begin_output_finalize();
916        assert!(
917            !state.should_interrupt_output(),
918            "a finalizing output must not be cut by the STATUS_END grace"
919        );
920        drop(guard);
921        assert!(
922            state.should_interrupt_output(),
923            "after the last finalize guard drops, the expired grace cuts again"
924        );
925    }
926
927    #[test]
928    fn abort_cuts_through_a_finalize_window() {
929        let state = state_with(STATUS_ABORT);
930        let _guard = state.begin_output_finalize();
931        assert!(
932            state.should_interrupt_output(),
933            "abort() is the hard cancel: it must cut even a finalizing output"
934        );
935    }
936
937    #[test]
938    fn running_scheduler_never_cuts_output() {
939        let state = state_with(STATUS_RUN);
940        assert!(!state.should_interrupt_output());
941    }
942
943    #[test]
944    fn overlapping_finalize_windows_hold_until_the_last_drops() {
945        let state = state_with(STATUS_END);
946        expire_grace(&state);
947        let g1 = state.begin_output_finalize();
948        let g2 = state.begin_output_finalize();
949        drop(g1);
950        assert!(
951            !state.should_interrupt_output(),
952            "one of two overlapping finalize windows dropping must keep the hold"
953        );
954        drop(g2);
955        assert!(state.should_interrupt_output());
956    }
957
958    #[test]
959    fn custom_io_poison_helpers_read_live_opaque_state() {
960        unsafe {
961            assert!(!input_custom_io_is_poisoned(null_mut()));
962            assert!(!output_custom_io_is_poisoned(null_mut()));
963
964            let input_opaque = Box::into_raw(Box::new(InputOpaque {
965                read: Box::new(|buf| buf.len() as i32),
966                seek: None,
967                poisoned: false,
968            }));
969            let input_avio = test_avio(input_opaque.cast(), 0);
970            assert!(!input_custom_io_is_poisoned(input_avio));
971            (*input_opaque).poisoned = true;
972            assert!(input_custom_io_is_poisoned(input_avio));
973            free_input_opaque(input_avio);
974
975            let output_opaque = Box::into_raw(Box::new(OutputOpaque {
976                write: Box::new(|buf| buf.len() as i32),
977                seek: None,
978                poisoned: false,
979            }));
980            let output_avio = test_avio(output_opaque.cast(), 1);
981            assert!(!output_custom_io_is_poisoned(output_avio));
982            (*output_opaque).poisoned = true;
983            assert!(output_custom_io_is_poisoned(output_avio));
984            free_output_opaque(output_avio);
985        }
986    }
987
988    #[test]
989    fn input_opaque_teardown_drops_each_callback_before_stable_repanic() {
990        let drops = Arc::new(AtomicUsize::new(0));
991        let read_bomb = CallbackDropBomb(Arc::clone(&drops));
992        let seek_bomb = CallbackDropBomb(Arc::clone(&drops));
993        let opaque = Box::into_raw(Box::new(InputOpaque {
994            read: Box::new(move |_buf| {
995                let _hold = &read_bomb;
996                0
997            }),
998            seek: Some(Box::new(move |_offset, _whence| {
999                let _hold = &seek_bomb;
1000                0
1001            })),
1002            poisoned: false,
1003        }));
1004        let avio_ctx = unsafe { test_avio(opaque.cast(), 0) };
1005
1006        let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| unsafe {
1007            free_input_opaque(avio_ctx)
1008        }))
1009        .expect_err("normal teardown must publish a stable panic after reclamation");
1010        assert_eq!(
1011            panic_message(payload.as_ref()),
1012            "custom-IO input callback state panicked during teardown"
1013        );
1014        assert_eq!(
1015            drops.load(Ordering::SeqCst),
1016            2,
1017            "read and seek callback boxes must both be disposed"
1018        );
1019    }
1020
1021    #[test]
1022    fn output_opaque_teardown_drops_each_callback_before_stable_repanic() {
1023        let drops = Arc::new(AtomicUsize::new(0));
1024        let write_bomb = CallbackDropBomb(Arc::clone(&drops));
1025        let seek_bomb = CallbackDropBomb(Arc::clone(&drops));
1026        let opaque = Box::into_raw(Box::new(OutputOpaque {
1027            write: Box::new(move |_buf| {
1028                let _hold = &write_bomb;
1029                0
1030            }),
1031            seek: Some(Box::new(move |_offset, _whence| {
1032                let _hold = &seek_bomb;
1033                0
1034            })),
1035            poisoned: false,
1036        }));
1037        let avio_ctx = unsafe { test_avio(opaque.cast(), 1) };
1038
1039        let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| unsafe {
1040            free_output_opaque(avio_ctx)
1041        }))
1042        .expect_err("normal teardown must publish a stable panic after reclamation");
1043        assert_eq!(
1044            panic_message(payload.as_ref()),
1045            "custom-IO output callback state panicked during teardown"
1046        );
1047        assert_eq!(
1048            drops.load(Ordering::SeqCst),
1049            2,
1050            "write and seek callback boxes must both be disposed"
1051        );
1052    }
1053
1054    #[test]
1055    fn output_format_teardown_preserves_the_deferred_stable_panic() {
1056        let drops = Arc::new(AtomicUsize::new(0));
1057        let write_bomb = CallbackDropBomb(Arc::clone(&drops));
1058        let opaque = Box::into_raw(Box::new(OutputOpaque {
1059            write: Box::new(move |_buf| {
1060                let _hold = &write_bomb;
1061                0
1062            }),
1063            seek: None,
1064            poisoned: false,
1065        }));
1066        let avio_ctx = unsafe { test_avio(opaque.cast(), 1) };
1067        let format_ctx = unsafe { ffmpeg_sys_next::avformat_alloc_context() };
1068        assert!(
1069            !format_ctx.is_null(),
1070            "test format context allocation failed"
1071        );
1072        unsafe {
1073            (*format_ctx).pb = avio_ctx;
1074        }
1075
1076        let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1077            out_fmt_ctx_free(format_ctx, true)
1078        }))
1079        .expect_err("the owner-level teardown must preserve the stable panic");
1080        assert_eq!(
1081            panic_message(payload.as_ref()),
1082            "custom-IO output callback state panicked during teardown"
1083        );
1084        assert_eq!(drops.load(Ordering::SeqCst), 1);
1085    }
1086
1087    #[test]
1088    fn custom_io_teardown_preserves_an_existing_unwind() {
1089        let input_drops = Arc::new(AtomicUsize::new(0));
1090        let read_bomb = CallbackDropBomb(Arc::clone(&input_drops));
1091        let seek_bomb = CallbackDropBomb(Arc::clone(&input_drops));
1092        let input_opaque = Box::into_raw(Box::new(InputOpaque {
1093            read: Box::new(move |_buf| {
1094                let _hold = &read_bomb;
1095                0
1096            }),
1097            seek: Some(Box::new(move |_offset, _whence| {
1098                let _hold = &seek_bomb;
1099                0
1100            })),
1101            poisoned: false,
1102        }));
1103        let input_avio = unsafe { test_avio(input_opaque.cast(), 0) };
1104        let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1105            let _free_on_unwind = FreeInputAvioOnDrop(input_avio);
1106            panic!("primary input unwind");
1107        }))
1108        .expect_err("the primary input panic must propagate");
1109        assert_eq!(panic_message(payload.as_ref()), "primary input unwind");
1110        assert_eq!(input_drops.load(Ordering::SeqCst), 2);
1111
1112        let output_drops = Arc::new(AtomicUsize::new(0));
1113        let write_bomb = CallbackDropBomb(Arc::clone(&output_drops));
1114        let seek_bomb = CallbackDropBomb(Arc::clone(&output_drops));
1115        let output_opaque = Box::into_raw(Box::new(OutputOpaque {
1116            write: Box::new(move |_buf| {
1117                let _hold = &write_bomb;
1118                0
1119            }),
1120            seek: Some(Box::new(move |_offset, _whence| {
1121                let _hold = &seek_bomb;
1122                0
1123            })),
1124            poisoned: false,
1125        }));
1126        let output_avio = unsafe { test_avio(output_opaque.cast(), 1) };
1127        let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1128            let _free_on_unwind = FreeOutputAvioOnDrop(output_avio);
1129            panic!("primary output unwind");
1130        }))
1131        .expect_err("the primary output panic must propagate");
1132        assert_eq!(panic_message(payload.as_ref()), "primary output unwind");
1133        assert_eq!(output_drops.load(Ordering::SeqCst), 2);
1134    }
1135}
ez_ffmpeg/core/context/mod.rs

ez_ffmpeg/core/context/
mod.rs
