diff --git a/CHANGELOG.md b/CHANGELOG.md index d70857d..3aa2ebd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -184,6 +184,14 @@ on `Call` (`get_call_participant_session_metrics`, - While a track is muted, its RTP clock keeps running, for audio and video. The first packet after an unmute shows the length of the mute (RFC 3550). Before, the timestamps continued from the last packet before the mute. +- `RemoteTrack::next_video_frame` decodes the VP9 SVC that browsers send in + flexible mode. Before, it gave the first frames and then no more frames: the + depacketizer kept the reference indices of earlier packets and rejected each + inter frame after the third reference index. +- `RemoteTrack::next_video_frame` asks the publisher for a keyframe when + packets arrive but no frame is decoded for 3 s, and again after each further + 3 s, as libwebrtc does. Before, a stream whose frames could not be assembled + stayed frozen and sent no keyframe request. # v0.1.0-preview.2 diff --git a/src/rtc/tracks/remote.rs b/src/rtc/tracks/remote.rs index ceb5d5c..c72def0 100644 --- a/src/rtc/tracks/remote.rs +++ b/src/rtc/tracks/remote.rs @@ -23,6 +23,7 @@ use std::sync::Mutex as StdMutex; use std::sync::{Arc, Weak}; use std::time::{Duration, Instant}; +use bytes::Bytes; use tokio::sync::{Mutex as AsyncMutex, Semaphore}; use webrtc::media::io::sample_builder::SampleBuilder; use webrtc::peer_connection::RTCPeerConnection; @@ -30,6 +31,7 @@ use webrtc::rtcp::packet::Packet as RtcpPacket; use webrtc::rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication; use webrtc::rtp::codecs::vp8::Vp8Packet; use webrtc::rtp::codecs::vp9::Vp9Packet; +use webrtc::rtp::packetizer::Depacketizer; use webrtc::track::track_remote::TrackRemote; use super::local::RtpPacket; @@ -54,6 +56,12 @@ const VIDEO_CLOCK_RATE: u32 = 90_000; /// one is in flight for at least a round trip, so asking faster only wastes /// uplink. const KEYFRAME_REQUEST_INTERVAL: Duration = Duration::from_secs(1); +/// Time without a decoded frame, while packets arrive, after which the publisher +/// is asked for a keyframe, and again after each further timeout. libwebrtc +/// `kMaxWaitForFrame`. +const VIDEO_STALL_TIMEOUT: Duration = Duration::from_secs(3); +/// Interval between `stream.rtc.remote.video_receive_stats` debug logs. +const VIDEO_STATS_INTERVAL: Duration = Duration::from_secs(5); /// The publishing participant a [`RemoteTrack`] belongs to. #[derive(Debug, Clone, Default, PartialEq)] @@ -143,7 +151,28 @@ pub struct Codec { /// decoder corrupts any frame that arrives out of order. enum VideoSamples { Vp8(SampleBuilder), - Vp9(SampleBuilder), + Vp9(SampleBuilder), +} + +/// [`Vp9Packet`] reset before each packet. `Vp9Packet::depacketize` keeps the +/// reference indices of earlier packets and fails once three accumulate, which +/// rejects every flexible-mode inter frame after the third. +#[derive(Default)] +struct Vp9Depacketizer(Vp9Packet); + +impl Depacketizer for Vp9Depacketizer { + fn depacketize(&mut self, packet: &Bytes) -> std::result::Result { + self.0 = Vp9Packet::default(); + self.0.depacketize(packet) + } + + fn is_partition_head(&self, payload: &Bytes) -> bool { + self.0.is_partition_head(payload) + } + + fn is_partition_tail(&self, marker: bool, payload: &Bytes) -> bool { + self.0.is_partition_tail(marker, payload) + } } impl VideoSamples { @@ -156,7 +185,7 @@ impl VideoSamples { )), VideoCodec::Vp9 => Self::Vp9(SampleBuilder::new( VIDEO_MAX_LATE, - Vp9Packet::default(), + Vp9Depacketizer::default(), VIDEO_CLOCK_RATE, )), } @@ -190,6 +219,28 @@ struct VideoDecode { decoder: VpxDecoder, ready: VecDeque, last_resolution: Option<(u32, u32)>, + /// The stream stalls if no frame is decoded before this time. + stall_deadline: Instant, + /// Payload type of the last packet, starting at the negotiated one. + payload_type: u8, + last_sequence_number: Option, + stats: VideoStats, + stats_since: Instant, +} + +/// Inbound video counts since the last stats log. [`SampleBuilder`] discards +/// its build errors, so packets without completed samples are the only sign of +/// a frame that could not be assembled. +#[derive(Default)] +struct VideoStats { + packets: usize, + /// Packets whose sequence number does not follow the previous packet's. + sequence_gaps: usize, + markers: usize, + samples: usize, + frames: usize, + decode_errors: usize, + last_payload_byte: Option, } /// Inbound audio decode state, plus frames already decoded but not yet handed to @@ -268,6 +319,14 @@ impl RemoteTrack { clock_rate: params.capability.clock_rate, channels: params.capability.channels, }; + tracing::debug!( + user_id = %participant.user_id, + ?track_type, + mime_type = %codec.mime_type, + payload_type = codec.payload_type, + ssrc = track.ssrc(), + "stream.rtc.remote.track_codec" + ); let decode = build_decoder(track_type, &codec); @@ -401,7 +460,8 @@ impl RemoteTrack { /// Packet loss and joining mid-stream both leave the decoder without a valid /// reference frame; this asks the publisher for a fresh keyframe (RTCP PLI, /// at most one per second) whenever that happens, so the stream recovers - /// instead of stalling silently. Concurrent calls are serialized, including + /// instead of stalling silently. It also asks once every 3 s while packets + /// arrive but no frame is decoded. Concurrent calls are serialized, including /// their native blocking decode work; do not mix this with raw reads. pub async fn next_video_frame(&self) -> Option { let Decode::Video(state) = &self.decode else { @@ -423,11 +483,15 @@ impl RemoteTrack { // Packet reordering/depacketization is cheap and stays on the async // task. Native VPx decode is measured in milliseconds at // 720p, so only complete samples cross into Tokio's blocking pool. - let samples = { + let (samples, stalled) = { let mut s = state.lock().unwrap_or_else(|e| e.into_inner()); - s.push_packet(packet) + let samples = s.push_packet(packet); + (samples, s.stall_timed_out(Instant::now())) }; if samples.is_empty() { + if stalled { + self.request_keyframe_throttled().await; + } continue; } let worker_permit = match Arc::clone(&self.video_decode_gate).acquire_owned().await { @@ -615,11 +679,54 @@ impl VideoDecode { /// Feed one RTP packet into the reassembler and return every sample it /// completes. This path performs no native decode work. fn push_packet(&mut self, packet: RtpPacket) -> Vec { + let ssrc = packet.header.ssrc; + let payload_type = packet.header.payload_type; + if payload_type != self.payload_type { + tracing::debug!( + ssrc, + from = self.payload_type, + to = payload_type, + "stream.rtc.remote.video_payload_type_changed" + ); + self.payload_type = payload_type; + } + self.stats.packets += 1; + let sequence_number = packet.header.sequence_number; + if self + .last_sequence_number + .is_some_and(|last| sequence_number != last.wrapping_add(1)) + { + self.stats.sequence_gaps += 1; + } + self.last_sequence_number = Some(sequence_number); + if packet.header.marker { + self.stats.markers += 1; + } + self.stats.last_payload_byte = packet.payload.first().copied(); + let mut completed = Vec::with_capacity(1); self.samples.push(packet); while let Some(sample) = self.samples.pop() { completed.push(sample); } + self.stats.samples += completed.len(); + + if self.stats_since.elapsed() >= VIDEO_STATS_INTERVAL { + let stats = std::mem::take(&mut self.stats); + tracing::debug!( + ssrc, + payload_type, + packets = stats.packets, + sequence_gaps = stats.sequence_gaps, + markers = stats.markers, + samples = stats.samples, + frames = stats.frames, + decode_errors = stats.decode_errors, + last_payload_byte = ?stats.last_payload_byte.map(|b| format!("{b:#04x}")), + "stream.rtc.remote.video_receive_stats" + ); + self.stats_since = Instant::now(); + } completed } @@ -640,6 +747,10 @@ impl VideoDecode { match self.decoder.decode(&sample.data, sample.packet_timestamp) { Ok(frames) => { + self.stats.frames += frames.len(); + if !frames.is_empty() { + self.stall_deadline = Instant::now() + VIDEO_STALL_TIMEOUT; + } for frame in frames { let resolution = (frame.width, frame.height); if self.last_resolution != Some(resolution) { @@ -654,6 +765,7 @@ impl VideoDecode { } } Err(e) => { + self.stats.decode_errors += 1; tracing::debug!(error = %e, "stream.rtc.remote.video_decode_failed"); needs_keyframe = true; } @@ -661,6 +773,16 @@ impl VideoDecode { } needs_keyframe } + + /// Whether a [`VIDEO_STALL_TIMEOUT`] without a decoded frame ended at `now`. + /// Each timeout is reported once; the next one starts at `now`. + fn stall_timed_out(&mut self, now: Instant) -> bool { + if now < self.stall_deadline { + return false; + } + self.stall_deadline = now + VIDEO_STALL_TIMEOUT; + true + } } impl Drop for RemoteTrack { @@ -713,6 +835,11 @@ fn build_decoder(track_type: TrackType, codec: &Codec) -> Decode { decoder, ready: VecDeque::new(), last_resolution: None, + stall_deadline: Instant::now() + VIDEO_STALL_TIMEOUT, + payload_type: codec.payload_type, + last_sequence_number: None, + stats: VideoStats::default(), + stats_since: Instant::now(), }))), Err(e) => { tracing::warn!( @@ -738,6 +865,7 @@ fn video_codec_for(mime_type: &str) -> Option { #[cfg(test)] mod tests { use super::*; + use crate::rtc::codecs::vpx::{EncodedFrame, VpxEncoder}; #[test] fn supported_video_mime_types_map_to_a_decoder() { @@ -746,6 +874,225 @@ mod tests { assert_eq!(video_codec_for("video/H264"), None); } + /// RTP packets for one VP9 frame in flexible mode, as browsers send VP9 + /// SVC: a 15-bit picture ID and, on an inter frame, one reference index. + fn vp9_flexible_mode_packets( + frame: &EncodedFrame, + picture_id: u16, + first_sequence_number: u16, + ) -> Vec { + let chunks: Vec<&[u8]> = frame.data.chunks(200).collect(); + let last = chunks.len() - 1; + chunks + .into_iter() + .enumerate() + .map(|(index, chunk)| { + // I (picture ID) + F (flexible mode). + let mut payload = vec![0x90, 0x80 | (picture_id >> 8) as u8, picture_id as u8]; + if !frame.key { + payload[0] |= 0x40; + // P_DIFF = 1, N = 0. + payload.push(1 << 1); + } + if index == 0 { + payload[0] |= 0x08; + } + if index == last { + payload[0] |= 0x04; + } + payload.extend_from_slice(chunk); + RtpPacket { + header: webrtc::rtp::header::Header { + version: 2, + payload_type: 98, + sequence_number: first_sequence_number.wrapping_add(index as u16), + timestamp: u32::from(picture_id) * 3_000, + marker: index == last, + ..Default::default() + }, + payload: Bytes::from(payload), + } + }) + .collect() + } + + #[test] + fn every_frame_of_a_flexible_mode_vp9_stream_is_decoded() { + let (width, height) = (160, 120); + let mut encoder = VpxEncoder::new(VpxCodec::Vp9, width, height, 400).expect("vp9 encoder"); + let image = vec![128u8; (width * height * 3 / 2) as usize]; + let state = vp9_decode(); + let mut state = state.lock().expect("video decode state"); + + let mut pictures: u16 = 0; + let mut sequence_number: u16 = 0; + for index in 0..10 { + let encoded = encoder + .encode(&image, index * 33, 33, index == 0) + .expect("encode"); + for frame in &encoded { + let packets = vp9_flexible_mode_packets(frame, pictures, sequence_number); + pictures += 1; + sequence_number = sequence_number.wrapping_add(packets.len() as u16); + for packet in packets { + let samples = state.push_packet(packet); + state.decode_samples(samples); + } + } + } + + // The last frame completes only when a packet of a later frame arrives. + assert!(pictures > 3, "the encoder must emit inter frames"); + assert_eq!(state.ready.len(), usize::from(pictures) - 1); + } + + fn vp9_decode() -> Arc> { + let codec = Codec { + mime_type: "video/VP9".to_owned(), + payload_type: 98, + clock_rate: VIDEO_CLOCK_RATE, + channels: 0, + }; + let Decode::Video(state) = build_decoder(TrackType::Video, &codec) else { + panic!("VP9 must have a decoder"); + }; + state + } + + #[test] + fn packets_that_complete_no_frame_stall_the_stream_after_the_timeout() { + let state = vp9_decode(); + let mut state = state.lock().expect("video decode state"); + + // One frame that never ends: no marker and no later timestamp. + for sequence_number in 0..10 { + state.push_packet(RtpPacket { + header: webrtc::rtp::header::Header { + version: 2, + payload_type: 98, + sequence_number, + ..Default::default() + }, + payload: Bytes::from_static(&[0x08, 0x00]), + }); + } + + assert!(!state.stall_timed_out(Instant::now())); + assert!(state.stall_timed_out(Instant::now() + VIDEO_STALL_TIMEOUT)); + } + + #[test] + fn a_stall_times_out_again_only_after_another_timeout() { + let state = vp9_decode(); + let mut state = state.lock().expect("video decode state"); + let first = Instant::now() + VIDEO_STALL_TIMEOUT; + + assert!(state.stall_timed_out(first)); + assert!(!state.stall_timed_out(first + VIDEO_STALL_TIMEOUT - Duration::from_millis(1))); + assert!(state.stall_timed_out(first + VIDEO_STALL_TIMEOUT)); + } + + #[test] + fn a_decoded_frame_restarts_the_stall_timeout() { + let (width, height) = (160, 120); + let mut encoder = VpxEncoder::new(VpxCodec::Vp9, width, height, 400).expect("vp9 encoder"); + let image = vec![128u8; (width * height * 3 / 2) as usize]; + let state = vp9_decode(); + let mut state = state.lock().expect("video decode state"); + + let before_decode = Instant::now(); + let mut sequence_number: u16 = 0; + // The keyframe completes when the first packet of the second frame arrives. + for picture_id in 0..2 { + let encoded = encoder + .encode(&image, i64::from(picture_id) * 33, 33, picture_id == 0) + .expect("encode"); + let packets = vp9_flexible_mode_packets(&encoded[0], picture_id, sequence_number); + sequence_number = sequence_number.wrapping_add(packets.len() as u16); + for packet in packets { + let samples = state.push_packet(packet); + state.decode_samples(samples); + } + } + + assert_eq!(state.ready.len(), 1); + assert!(!state.stall_timed_out(before_decode + VIDEO_STALL_TIMEOUT)); + } + + #[tokio::test] + async fn a_track_whose_packets_complete_no_frame_requests_a_keyframe() { + use crate::rtc::peer; + use webrtc::rtp_transceiver::rtp_codec::{RTCRtpCodecCapability, RTPCodecType}; + use webrtc::track::track_local::TrackLocalWriter; + use webrtc::track::track_local::track_local_static_rtp::TrackLocalStaticRTP; + + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + let local = Arc::new(TrackLocalStaticRTP::new( + RTCRtpCodecCapability { + mime_type: "video/VP9".to_owned(), + clock_rate: VIDEO_CLOCK_RATE, + sdp_fmtp_line: "profile-id=0".to_owned(), + ..Default::default() + }, + "video".to_owned(), + "stalled".to_owned(), + )); + let rtp_sender = sender.add_track(local.clone()).await.expect("add track"); + let (receiver, mut remote_rx) = peer::connect_receiver(&sender, RTPCodecType::Video).await; + + // One frame that never ends: no marker and no later timestamp. + let writer = tokio::spawn(async move { + let mut sequence_number: u16 = 0; + loop { + let packet = RtpPacket { + header: webrtc::rtp::header::Header { + version: 2, + sequence_number, + ..Default::default() + }, + payload: Bytes::from_static(&[0x08, 0x00]), + }; + if local.write_rtp(&packet).await.is_err() { + return; + } + sequence_number = sequence_number.wrapping_add(1); + tokio::time::sleep(Duration::from_millis(33)).await; + } + }); + let track = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let started = Instant::now(); + let remote = RemoteTrack::from_webrtc(track, TrackType::Video, &receiver); + let reader = tokio::spawn(async move { remote.next_video_frame().await }); + + let requested_after = tokio::time::timeout(VIDEO_STALL_TIMEOUT * 3, async { + loop { + let (packets, _) = rtp_sender.read_rtcp().await.expect("read RTCP"); + if packets.iter().any(|packet| { + packet + .as_any() + .downcast_ref::() + .is_some() + }) { + return started.elapsed(); + } + } + }) + .await; + writer.abort(); + reader.abort(); + let _ = receiver.close().await; + let _ = sender.close().await; + + let requested_after = requested_after.expect("a keyframe request"); + assert!( + requested_after >= VIDEO_STALL_TIMEOUT, + "the keyframe request came after {requested_after:?}" + ); + } + /// One 20 ms frame of 440 Hz tone, loud enough that a silent or badly /// rebuilt frame is obvious. fn tone_20ms() -> Vec {