Compare revisions

Arun Raghavan · Sebastian Dröge · Sebastian Dröge · Sebastian Dröge · Sebastian Dröge · Sebastian Dröge
--- a/net/rtp/src/av1/depay/imp.rs
+++ b/net/rtp/src/av1/depay/imp.rs
@@ -553,7 +553,10 @@ impl RTPAv1Depay {
            first = false;

            // ignore these OBU types
-            if matches!(obu.obu_type, ObuType::TemporalDelimiter | ObuType::TileList) {
+            if matches!(
+                obu.obu_type,
+                ObuType::TemporalDelimiter | ObuType::TileList | ObuType::Padding
+            ) {
                gst::trace!(CAT, imp: self, "Dropping {:?} of size {element_size}", obu.obu_type);
                reader
                    .seek(SeekFrom::Current(element_size as i64))

--- a/net/rtp/src/av1/depay/tests.rs
+++ b/net/rtp/src/av1/depay/tests.rs
@@ -35,7 +35,7 @@ fn test_depayloader() {
        ), ( // 2 OBUs, last is fragmented
            vec![
                0b0110_0000,
-                0b0000_0110, 0b0111_1000, 1, 2, 3, 4, 5,
+                0b0000_0110, 0b0011_0000, 1, 2, 3, 4, 5,
                             0b0011_0000, 1, 2, 3,
            ],
            gst::ClockTime::from_seconds(1),
@@ -68,7 +68,7 @@ fn test_depayloader() {
        ),
        (
            gst::ClockTime::from_seconds(1),
-            vec![0b0001_0010, 0, 0b0111_1010, 0b0000_0101, 1, 2, 3, 4, 5],
+            vec![0b0001_0010, 0, 0b0011_0010, 0b0000_0101, 1, 2, 3, 4, 5],
        ),
        (
            gst::ClockTime::from_seconds(1),

--- a/net/rtp/src/av1/pay/imp.rs
+++ b/net/rtp/src/av1/pay/imp.rs
@@ -35,6 +35,7 @@ static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
 struct PacketOBUData {
    obu_count: usize,
    payload_size: u32,
+    start_of_coded_video_sequence: bool,
    last_obu_fragment_size: Option<u32>,
    omit_last_size_field: bool,
    ends_temporal_unit: bool,
@@ -45,6 +46,7 @@ impl Default for PacketOBUData {
        PacketOBUData {
            payload_size: 1, // 1 byte is used for the aggregation header
            omit_last_size_field: true,
+            start_of_coded_video_sequence: false,
            obu_count: 0,
            last_obu_fragment_size: None,
            ends_temporal_unit: false,
@@ -55,12 +57,13 @@ impl Default for PacketOBUData {
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 struct ObuData {
    info: SizedObu,
+    keyframe: bool,
    bytes: Vec<u8>,
    offset: usize,
    id: u64,
 }

-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
 struct State {
    /// Holds header information and raw bytes for all received OBUs,
    /// as well as DTS and PTS
@@ -70,10 +73,6 @@ struct State {
    /// left over from the previous RTP packet
    open_obu_fragment: bool,

-    /// Indicates the next constructed packet will be the first in its sequence
-    /// (Corresponds to `N` field in the aggregation header)
-    first_packet_in_seq: bool,
-
    /// If the input is TU or frame aligned.
    framed: bool,
 }
@@ -83,17 +82,6 @@ pub struct RTPAv1Pay {
    state: AtomicRefCell<State>,
 }

-impl Default for State {
-    fn default() -> Self {
-        Self {
-            obus: VecDeque::new(),
-            open_obu_fragment: false,
-            first_packet_in_seq: true,
-            framed: false,
-        }
-    }
-}
-
 impl RTPAv1Pay {
    fn reset(&self, state: &mut State, full: bool) {
        gst::debug!(CAT, imp: self, "resetting state");
@@ -115,6 +103,7 @@ impl RTPAv1Pay {
        state: &mut State,
        id: u64,
        data: &[u8],
+        keyframe: bool,
        marker: bool,
    ) -> Result<gst::FlowSuccess, gst::FlowError> {
        let mut reader = Cursor::new(data);
@@ -127,13 +116,11 @@ impl RTPAv1Pay {
            // tile lists and temporal delimiters should not be transmitted,
            // see section 5 of the RTP AV1 spec
            match obu.obu_type {
-                // completely ignore tile lists
-                ObuType::TileList => {
-                    gst::log!(CAT, imp: self, "ignoring tile list OBU");
+                // completely ignore tile lists and padding
+                ObuType::TileList | ObuType::Padding => {
+                    gst::log!(CAT, imp: self, "ignoring {:?} OBU", obu.obu_type);
                    reader
-                        .seek(SeekFrom::Current(
-                            (obu.header_len + obu.leb_size + obu.size) as i64,
-                        ))
+                        .seek(SeekFrom::Current(obu.size as i64))
                        .map_err(err_flow!(self, buf_read))?;
                }

@@ -149,6 +136,7 @@ impl RTPAv1Pay {
                    }
                    state.obus.push_back(ObuData {
                        info: obu,
+                        keyframe,
                        bytes: Vec::new(),
                        offset: 0,
                        id,
@@ -180,6 +168,7 @@ impl RTPAv1Pay {

                    state.obus.push_back(ObuData {
                        info: obu,
+                        keyframe,
                        bytes,
                        offset: 0,
                        id,
@@ -226,6 +215,11 @@ impl RTPAv1Pay {
        let mut pending_bytes = 0;
        let mut required_ids = None::<(u8, u8)>;

+        // Detect if this packet starts a keyframe and contains a sequence header, and if so
+        // set the N flag to indicate that this is the start of a new codec video sequence.
+        let mut contains_keyframe = false;
+        let mut contains_sequence_header = false;
+
        // figure out how many OBUs we can fit into this packet
        for (idx, obu) in state.obus.iter().enumerate() {
            // for OBUs with extension headers, spatial and temporal IDs must be equal
@@ -256,6 +250,8 @@ impl RTPAv1Pay {
                        );
                    }

+                    packet.start_of_coded_video_sequence =
+                        contains_keyframe && contains_sequence_header;
                    packet.ends_temporal_unit = true;
                    if packet.obu_count > 3 {
                        packet.payload_size += pending_bytes;
@@ -265,6 +261,7 @@ impl RTPAv1Pay {
                    return Some(packet);
                }

+                contains_keyframe |= obu.keyframe;
                continue;
            } else if packet.payload_size >= payload_limit
                || (packet.obu_count > 0 && current.obu_type == ObuType::SequenceHeader)
@@ -274,6 +271,8 @@ impl RTPAv1Pay {
                    packet.payload_size += pending_bytes;
                    packet.omit_last_size_field = false;
                }
+                packet.start_of_coded_video_sequence =
+                    contains_keyframe && contains_sequence_header;
                packet.ends_temporal_unit = marker && idx == state.obus.len() - 1;
                return Some(packet);
            }
@@ -282,6 +281,8 @@ impl RTPAv1Pay {
            if packet.payload_size + pending_bytes + current.full_size() <= payload_limit {
                packet.obu_count += 1;
                packet.payload_size += current.partial_size() + pending_bytes;
+                contains_keyframe |= obu.keyframe;
+                contains_sequence_header |= obu.info.obu_type == ObuType::SequenceHeader;
                pending_bytes = current.leb_size;
            }
            // would it fit without the size field?
@@ -290,6 +291,10 @@ impl RTPAv1Pay {
            {
                packet.obu_count += 1;
                packet.payload_size += current.partial_size() + pending_bytes;
+                contains_keyframe |= obu.keyframe;
+                contains_sequence_header |= obu.info.obu_type == ObuType::SequenceHeader;
+                packet.start_of_coded_video_sequence =
+                    contains_keyframe && contains_sequence_header;
                packet.ends_temporal_unit = marker && idx == state.obus.len() - 1;

                return Some(packet);
@@ -313,11 +318,16 @@ impl RTPAv1Pay {
                        Some(payload_limit - packet.payload_size - pending_bytes - leb_size);
                    packet.payload_size = payload_limit;
                    packet.omit_last_size_field = leb_size == 0;
+                    contains_keyframe |= obu.keyframe;
+                    contains_sequence_header |= obu.info.obu_type == ObuType::SequenceHeader;
                } else if packet.obu_count > 3 {
                    packet.ends_temporal_unit = marker && idx == state.obus.len() - 1;
                    packet.payload_size += pending_bytes;
                }

+                packet.start_of_coded_video_sequence =
+                    contains_keyframe && contains_sequence_header;
+
                return Some(packet);
            }
        }
@@ -327,6 +337,7 @@ impl RTPAv1Pay {
                packet.payload_size += pending_bytes;
                packet.omit_last_size_field = false;
            }
+            packet.start_of_coded_video_sequence = contains_keyframe && contains_sequence_header;
            packet.ends_temporal_unit = true;

            Some(packet)
@@ -363,17 +374,15 @@ impl RTPAv1Pay {
            };

            let aggr_header: [u8; 1] = [
-                    (state.open_obu_fragment as u8) << 7 |                  // Z
-                    ((packet.last_obu_fragment_size.is_some()) as u8) << 6 |  // Y
-                    (w as u8) << 4 |                                        // W
-                    (state.first_packet_in_seq as u8) << 3                  // N
+                    (state.open_obu_fragment as u8) << 7 |                       // Z
+                    ((packet.last_obu_fragment_size.is_some()) as u8) << 6 |     // Y
+                    (w as u8) << 4 |                                             // W
+                    (packet.start_of_coded_video_sequence as u8) << 3 // N
                ; 1];

            writer
                .write(&aggr_header)
                .map_err(err_flow!(self, aggr_header_write))?;
-
-            state.first_packet_in_seq = false;
        }

        let mut start_id = None;
@@ -600,9 +609,10 @@ impl crate::basepay::RtpBasePay2Impl for RTPAv1Pay {
            gst::FlowError::Error
        })?;

+        let keyframe = !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT);
        // Does the buffer finished a full TU?
        let marker = buffer.flags().contains(gst::BufferFlags::MARKER) || state.framed;
-        let res = self.handle_new_obus(&mut state, id, map.as_slice(), marker)?;
+        let res = self.handle_new_obus(&mut state, id, map.as_slice(), keyframe, marker)?;
        drop(map);
        drop(state);

@@ -656,7 +666,7 @@ mod tests {
                    obus: VecDeque::from(vec![
                        ObuData {
                            info: SizedObu {
-                                obu_type: ObuType::Padding,
+                                obu_type: ObuType::Frame,
                                size: 3,
                                ..base_obu
                            },
@@ -713,6 +723,15 @@ mod tests {
                                size: 0,
                                ..base_obu
                            },
+                            keyframe: true,
+                            ..ObuData::default()
+                        },
+                        ObuData {
+                            info: SizedObu {
+                                obu_type: ObuType::SequenceHeader,
+                                size: 0,
+                                ..base_obu
+                            },
                            ..ObuData::default()
                        },
                        ObuData {
@@ -726,7 +745,7 @@ mod tests {
                        },
                        ObuData {
                            info: SizedObu {
-                                obu_type: ObuType::Padding,
+                                obu_type: ObuType::Frame,
                                size: 6,
                                ..base_obu
                            },
@@ -787,6 +806,7 @@ mod tests {
                Some(PacketOBUData {
                    obu_count: 3,
                    payload_size: 18,
+                    start_of_coded_video_sequence: false,
                    last_obu_fragment_size: None,
                    omit_last_size_field: true,
                    ends_temporal_unit: true,
@@ -803,8 +823,9 @@ mod tests {
            ),
            (
                Some(PacketOBUData {
-                    obu_count: 4,
-                    payload_size: 34,
+                    obu_count: 5,
+                    payload_size: 36,
+                    start_of_coded_video_sequence: true,
                    last_obu_fragment_size: None,
                    omit_last_size_field: false,
                    ends_temporal_unit: true,
@@ -849,16 +870,13 @@ mod tests {
                state
                    .obus
                    .iter()
-                    .filter(|o| o.info.obu_type != ObuType::TemporalDelimiter)
+                    .filter(|o| o.info.obu_type != ObuType::TemporalDelimiter
+                        && o.info.obu_type != ObuType::Padding)
                    .cloned()
                    .collect::<Vec<_>>(),
                results[idx].1.obus.iter().cloned().collect::<Vec<_>>()
            );
            assert_eq!(state.open_obu_fragment, results[idx].1.open_obu_fragment);
-            assert_eq!(
-                state.first_packet_in_seq,
-                results[idx].1.first_packet_in_seq
-            );
        }
    }
 }
--- a/net/rtp/src/av1/pay/tests.rs
+++ b/net/rtp/src/av1/pay/tests.rs
@@ -26,15 +26,16 @@ fn test_payloader() {
    let test_buffers: [(u64, Vec<u8>); 3] = [
        (
            0,
-            vec![   // this should result in exactly 25 bytes for the RTP payload
+            vec![   // this should result in exactly 27 bytes for the RTP payload
                0b0001_0010, 0,
+                0b0000_1010, 0,
                0b0011_0010, 0b0000_1100, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12,
                0b0011_0010, 0b0000_1001, 1, 2, 3, 4, 5, 6, 7, 8, 9,
            ],
        ), (
            0,
            vec![   // these all have to go in separate packets since their IDs mismatch
-                0b0111_1010, 0b0000_0100, 1, 2, 3, 4,
+                0b0011_0010, 0b0000_0100, 1, 2, 3, 4,
                0b0011_0110, 0b0010_1000, 0b0000_0101, 1, 2, 3, 4, 5,
                0b0011_0110, 0b0100_1000, 0b0000_0001, 1,
            ],
@@ -53,7 +54,8 @@ fn test_payloader() {
            false,  // marker bit
            0,      // relative RTP timestamp
            vec![   // payload bytes
-                0b0010_1000,
+                0b0011_1000,
+                0b0000_0001, 0b0000_1000,
                0b0000_1101, 0b0011_0000, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12,
                             0b0011_0000, 1, 2, 3, 4, 5, 6, 7, 8, 9,
            ],
@@ -62,7 +64,7 @@ fn test_payloader() {
            0,
            vec![
                0b0001_0000,
-                0b0111_1000, 1, 2, 3, 4,
+                0b0011_0000, 1, 2, 3, 4,
            ]
        ), (
            false,
@@ -95,7 +97,7 @@ fn test_payloader() {
        let pay = h.element().unwrap();
        pay.set_property(
            "mtu",
-            25u32 + rtp_types::RtpPacket::MIN_RTP_PACKET_LEN as u32,
+            27u32 + rtp_types::RtpPacket::MIN_RTP_PACKET_LEN as u32,
        );
    }
    h.play();
No results found