fmp4mux: Add support for sub-fragments / chunking

Allow outputting sub-fragments (chunks in CMAF terms) that are shorter
than the fragment duration and don't usually start on a keyframe. By
this the buffering requirements of the element is reduced to one chunk
duration, as is the latency.

This is used for formats like low-latency / LL-HLS and DASH.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1059>

docs/plugins/gst_plugins_cache.json

@@ -1728,6 +1728,20 @@
                 ],
                 "kind": "object",
                 "properties": {
+                    "chunk-duration": {
+                        "blurb": "Duration for each FMP4 chunk (default = no chunks)",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": false,
+                        "controllable": false,
+                        "default": "18446744073709551615",
+                        "max": "18446744073709551615",
+                        "min": "0",
+                        "mutable": "ready",
+                        "readable": true,
+                        "type": "guint64",
+                        "writable": true
+                    },
                     "fragment-duration": {
                         "blurb": "Duration for each FMP4 fragment",
                         "conditionally-available": false,

mux/fmp4/src/fmp4mux/boxes.rs

@@ -1573,19 +1573,22 @@ pub(super) fn create_fmp4_fragment_header(
 ) -> Result<(gst::Buffer, u64), Error> {
     let mut v = vec![];
 
-    let (brand, compatible_brands) =
-        brands_from_variant_and_caps(cfg.variant, cfg.streams.iter().map(|s| &s.caps));
+    // Don't write a `styp` if this is only a chunk.
+    if !cfg.chunk {
+        let (brand, compatible_brands) =
+            brands_from_variant_and_caps(cfg.variant, cfg.streams.iter().map(|s| &s.caps));
 
-    write_box(&mut v, b"styp", |v| {
-        // major brand
-        v.extend(brand);
-        // minor version
-        v.extend(0u32.to_be_bytes());
-        // compatible brands
-        v.extend(compatible_brands.into_iter().flatten());
+        write_box(&mut v, b"styp", |v| {
+            // major brand
+            v.extend(brand);
+            // minor version
+            v.extend(0u32.to_be_bytes());
+            // compatible brands
+            v.extend(compatible_brands.into_iter().flatten());
 
-        Ok(())
-    })?;
+            Ok(())
+        })?;
+    }
 
     let styp_len = v.len();
 

mux/fmp4/src/fmp4mux/mod.rs

@@ -110,6 +110,9 @@ pub(crate) struct FragmentHeaderConfiguration<'a> {
     /// Sequence number for this fragment.
     sequence_number: u32,
 
+    /// If this is a full fragment or only a chunk.
+    chunk: bool,
+
     streams: &'a [FragmentHeaderStream],
     buffers: &'a [Buffer],
 }

mux/fmp4/tests/tests.rs

@@ -1,3 +1,4 @@
+// Copyright (C) 2021 Sebastian Dröge <sebastian@centricular.com>
 //
 // This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
 // If a copy of the MPL was not distributed with this file, You can obtain one at
@@ -1285,3 +1286,386 @@ fn test_buffer_multi_stream_short_gops() {
     let ev = h1.pull_event().unwrap();
     assert_eq!(ev.type_(), gst::EventType::Eos);
 }
+
+#[test]
+fn test_chunking_single_stream() {
+    init();
+
+    let caps = gst::Caps::builder("video/x-h264")
+        .field("width", 1920i32)
+        .field("height", 1080i32)
+        .field("framerate", gst::Fraction::new(30, 1))
+        .field("stream-format", "avc")
+        .field("alignment", "au")
+        .field("codec_data", gst::Buffer::with_size(1).unwrap())
+        .build();
+
+    let mut h = gst_check::Harness::new("cmafmux");
+
+    // 5s fragment duration, 1s chunk duration
+    h.element()
+        .unwrap()
+        .set_property("fragment-duration", 5.seconds());
+    h.element()
+        .unwrap()
+        .set_property("chunk-duration", 1.seconds());
+
+    h.set_src_caps(caps);
+    h.play();
+
+    // Push 15 buffers of 0.5s each, 1st and 11th buffer without DELTA_UNIT flag
+    for i in 0..15 {
+        let mut buffer = gst::Buffer::with_size(1).unwrap();
+        {
+            let buffer = buffer.get_mut().unwrap();
+            buffer.set_pts(i * 500.mseconds());
+            buffer.set_dts(i * 500.mseconds());
+            buffer.set_duration(500.mseconds());
+            if i != 0 && i != 10 {
+                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
+            }
+        }
+        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));
+
+        if i == 2 {
+            let ev = loop {
+                let ev = h.pull_upstream_event().unwrap();
+                if ev.type_() != gst::EventType::Reconfigure
+                    && ev.type_() != gst::EventType::Latency
+                {
+                    break ev;
+                }
+            };
+
+            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
+            assert_eq!(
+                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
+                gst_video::UpstreamForceKeyUnitEvent {
+                    running_time: Some(5.seconds()),
+                    all_headers: true,
+                    count: 0
+                }
+            );
+        }
+    }
+
+    // Crank the clock: this should bring us to the end of the first fragment
+    h.crank_single_clock_wait().unwrap();
+
+    let header = h.pull().unwrap();
+    assert_eq!(
+        header.flags(),
+        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
+    );
+    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO));
+    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO));
+
+    // There should be 7 chunks now, and the 1st and 6th are starting a fragment.
+    // Each chunk should have two buffers.
+    for chunk in 0..7 {
+        let chunk_header = h.pull().unwrap();
+        if chunk == 0 || chunk == 5 {
+            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
+        } else {
+            assert_eq!(
+                chunk_header.flags(),
+                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
+            );
+        }
+        assert_eq!(chunk_header.pts(), Some(chunk * 1.seconds()));
+        assert_eq!(chunk_header.dts(), Some(chunk * 1.seconds()));
+        assert_eq!(chunk_header.duration(), Some(1.seconds()));
+
+        for buffer_idx in 0..2 {
+            let buffer = h.pull().unwrap();
+            if buffer_idx == 1 {
+                assert_eq!(
+                    buffer.flags(),
+                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
+                );
+            } else {
+                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
+            }
+            assert_eq!(
+                buffer.pts(),
+                Some((chunk * 2 + buffer_idx) * 500.mseconds())
+            );
+            assert_eq!(
+                buffer.dts(),
+                Some((chunk * 2 + buffer_idx) * 500.mseconds())
+            );
+            assert_eq!(buffer.duration(), Some(500.mseconds()));
+        }
+    }
+
+    h.push_event(gst::event::Eos::new());
+
+    // There should be the remaining chunk now, containing one 500ms buffer.
+    for chunk in 7..8 {
+        let chunk_header = h.pull().unwrap();
+        assert_eq!(
+            chunk_header.flags(),
+            gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
+        );
+        assert_eq!(chunk_header.pts(), Some(chunk * 1.seconds()));
+        assert_eq!(chunk_header.dts(), Some(chunk * 1.seconds()));
+        assert_eq!(chunk_header.duration(), Some(500.mseconds()));
+
+        for buffer_idx in 0..1 {
+            let buffer = h.pull().unwrap();
+            assert_eq!(
+                buffer.flags(),
+                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
+            );
+            assert_eq!(
+                buffer.pts(),
+                Some((chunk * 2 + buffer_idx) * 500.mseconds())
+            );
+            assert_eq!(
+                buffer.dts(),
+                Some((chunk * 2 + buffer_idx) * 500.mseconds())
+            );
+            assert_eq!(buffer.duration(), Some(500.mseconds()));
+        }
+    }
+
+    let ev = h.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::StreamStart);
+    let ev = h.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::Caps);
+    let ev = h.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::Segment);
+    let ev = h.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::Eos);
+}
+
+#[test]
+fn test_chunking_multi_stream() {
+    init();
+
+    let mut h1 = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));
+    let mut h2 = gst_check::Harness::with_element(&h1.element().unwrap(), Some("sink_1"), None);
+
+    // 5s fragment duration, 1s chunk duration
+    h1.element()
+        .unwrap()
+        .set_property("fragment-duration", 5.seconds());
+    h1.element()
+        .unwrap()
+        .set_property("chunk-duration", 1.seconds());
+
+    h1.set_src_caps(
+        gst::Caps::builder("video/x-h264")
+            .field("width", 1920i32)
+            .field("height", 1080i32)
+            .field("framerate", gst::Fraction::new(30, 1))
+            .field("stream-format", "avc")
+            .field("alignment", "au")
+            .field("codec_data", gst::Buffer::with_size(1).unwrap())
+            .build(),
+    );
+    h1.play();
+
+    h2.set_src_caps(
+        gst::Caps::builder("audio/mpeg")
+            .field("mpegversion", 4i32)
+            .field("channels", 1i32)
+            .field("rate", 44100i32)
+            .field("stream-format", "raw")
+            .field("base-profile", "lc")
+            .field("profile", "lc")
+            .field("level", "2")
+            .field(
+                "codec_data",
+                gst::Buffer::from_slice([0x12, 0x08, 0x56, 0xe5, 0x00]),
+            )
+            .build(),
+    );
+    h2.play();
+
+    let output_offset = (60 * 60 * 1000).seconds();
+
+    // Push 15 buffers of 0.5s each, 1st and 11th buffer without DELTA_UNIT flag
+    for i in 0..15 {
+        let mut buffer = gst::Buffer::with_size(1).unwrap();
+        {
+            let buffer = buffer.get_mut().unwrap();
+            buffer.set_pts(i * 500.mseconds());
+            buffer.set_dts(i * 500.mseconds());
+            buffer.set_duration(500.mseconds());
+            if i != 0 && i != 10 {
+                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
+            }
+        }
+        assert_eq!(h1.push(buffer), Ok(gst::FlowSuccess::Ok));
+
+        let mut buffer = gst::Buffer::with_size(1).unwrap();
+        {
+            let buffer = buffer.get_mut().unwrap();
+            buffer.set_pts(i * 500.mseconds());
+            buffer.set_dts(i * 500.mseconds());
+            buffer.set_duration(500.mseconds());
+        }
+        assert_eq!(h2.push(buffer), Ok(gst::FlowSuccess::Ok));
+
+        if i == 2 {
+            let ev = loop {
+                let ev = h1.pull_upstream_event().unwrap();
+                if ev.type_() != gst::EventType::Reconfigure
+                    && ev.type_() != gst::EventType::Latency
+                {
+                    break ev;
+                }
+            };
+
+            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
+            assert_eq!(
+                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
+                gst_video::UpstreamForceKeyUnitEvent {
+                    running_time: Some(5.seconds()),
+                    all_headers: true,
+                    count: 0
+                }
+            );
+
+            let ev = loop {
+                let ev = h2.pull_upstream_event().unwrap();
+                if ev.type_() != gst::EventType::Reconfigure
+                    && ev.type_() != gst::EventType::Latency
+                {
+                    break ev;
+                }
+            };
+
+            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
+            assert_eq!(
+                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
+                gst_video::UpstreamForceKeyUnitEvent {
+                    running_time: Some(5.seconds()),
+                    all_headers: true,
+                    count: 0
+                }
+            );
+        }
+    }
+
+    // Crank the clock: this should bring us to the end of the first fragment
+    h1.crank_single_clock_wait().unwrap();
+
+    let header = h1.pull().unwrap();
+    assert_eq!(
+        header.flags(),
+        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
+    );
+    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
+    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));
+
+    // There should be 7 chunks now, and the 1st and 6th are starting a fragment.
+    // Each chunk should have two buffers.
+    for chunk in 0..7 {
+        let chunk_header = h1.pull().unwrap();
+        if chunk == 0 || chunk == 5 {
+            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
+        } else {
+            assert_eq!(
+                chunk_header.flags(),
+                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
+            );
+        }
+        assert_eq!(
+            chunk_header.pts(),
+            Some(chunk * 1.seconds() + output_offset)
+        );
+        assert_eq!(
+            chunk_header.dts(),
+            Some(chunk * 1.seconds() + output_offset)
+        );
+        assert_eq!(chunk_header.duration(), Some(1.seconds()));
+
+        for buffer_idx in 0..2 {
+            for stream_idx in 0..2 {
+                let buffer = h1.pull().unwrap();
+                if buffer_idx == 1 && stream_idx == 1 {
+                    assert_eq!(
+                        buffer.flags(),
+                        gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
+                    );
+                } else {
+                    assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
+                }
+                assert_eq!(
+                    buffer.pts(),
+                    Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
+                );
+
+                if stream_idx == 0 {
+                    assert_eq!(
+                        buffer.dts(),
+                        Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
+                    );
+                } else {
+                    assert!(buffer.dts().is_none());
+                }
+                assert_eq!(buffer.duration(), Some(500.mseconds()));
+            }
+        }
+    }
+
+    h1.push_event(gst::event::Eos::new());
+    h2.push_event(gst::event::Eos::new());
+
+    // There should be the remaining chunk now, containing one 500ms buffer.
+    for chunk in 7..8 {
+        let chunk_header = h1.pull().unwrap();
+        assert_eq!(
+            chunk_header.flags(),
+            gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
+        );
+        assert_eq!(
+            chunk_header.pts(),
+            Some(chunk * 1.seconds() + output_offset)
+        );
+        assert_eq!(
+            chunk_header.dts(),
+            Some(chunk * 1.seconds() + output_offset)
+        );
+        assert_eq!(chunk_header.duration(), Some(500.mseconds()));
+
+        for buffer_idx in 0..1 {
+            for stream_idx in 0..2 {
+                let buffer = h1.pull().unwrap();
+                if buffer_idx == 0 && stream_idx == 1 {
+                    assert_eq!(
+                        buffer.flags(),
+                        gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
+                    );
+                } else {
+                    assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
+                }
+
+                assert_eq!(
+                    buffer.pts(),
+                    Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
+                );
+                if stream_idx == 0 {
+                    assert_eq!(
+                        buffer.dts(),
+                        Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
+                    );
+                } else {
+                    assert!(buffer.dts().is_none());
+                }
+                assert_eq!(buffer.duration(), Some(500.mseconds()));
+            }
+        }
+    }
+
+    let ev = h1.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::StreamStart);
+    let ev = h1.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::Caps);
+    let ev = h1.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::Segment);
+    let ev = h1.pull_event().unwrap();
+    assert_eq!(ev.type_(), gst::EventType::Eos);
+}