diff --git a/.changeset/h265_length_prefixed_ingest.md b/.changeset/h265_length_prefixed_ingest.md new file mode 100644 index 000000000..3c964c672 --- /dev/null +++ b/.changeset/h265_length_prefixed_ingest.md @@ -0,0 +1,7 @@ +--- +livekit: patch +livekit-capture: patch +livekit-ffi: patch +--- + +Ingest length-prefixed (`avc3`/`hev1`) pre-encoded H.264 and H.265 with `LengthPrefixedAccessUnitParser`, `access_unit_from_h265_hvcc` and `hev1` caps in the GStreamer source, read LKTS packet-trailer SEI with `frame_metadata_from_sei`, and announce the codec of single-layer video tracks to the server. diff --git a/.changeset/preencoded_pacer.md b/.changeset/preencoded_pacer.md new file mode 100644 index 000000000..05767b603 --- /dev/null +++ b/.changeset/preencoded_pacer.md @@ -0,0 +1,9 @@ +--- +webrtc-sys: patch +libwebrtc: patch +livekit: patch +livekit-capture: patch +livekit-ffi: patch +--- + +Read WebRTC field trials from `LIVEKIT_WEBRTC_FIELD_TRIALS` (for example `WebRTC-Video-Pacing/factor:10/`) and stop capping the starting bandwidth estimate of pre-encoded tracks, so their packets no longer queue in the pacer after publishing. diff --git a/.changeset/preencoded_publish_path.md b/.changeset/preencoded_publish_path.md new file mode 100644 index 000000000..94aefeed4 --- /dev/null +++ b/.changeset/preencoded_publish_path.md @@ -0,0 +1,7 @@ +--- +livekit: patch +livekit-capture: patch +livekit-ffi: patch +--- + +Build length-prefixed access units without rescanning them for start codes, speed up Annex-B start-code search, and publish pre-encoded video as a single layer even when simulcast is requested. diff --git a/Cargo.lock b/Cargo.lock index b57eb4163..ed615d5aa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4094,6 +4094,7 @@ dependencies = [ "image", "livekit", "livekit-api", + "livekit-capture", "log", "metal 0.32.0", "nokhwa", diff --git a/examples/local_video/Cargo.toml b/examples/local_video/Cargo.toml index da1b5088e..dda21a419 100644 --- a/examples/local_video/Cargo.toml +++ b/examples/local_video/Cargo.toml @@ -17,6 +17,11 @@ name = "subscriber" path = "src/subscriber.rs" required-features = ["desktop"] +[[bin]] +name = "encoded_publisher" +path = "src/encoded_publisher.rs" +required-features = ["desktop"] + [[bin]] name = "list_devices" path = "src/list_devices.rs" @@ -32,6 +37,7 @@ tokio = { workspace = true, features = ["full", "parking_lot"] } tokio-stream = { workspace = true } livekit = { workspace = true, features = ["rustls-tls-native-roots"] } livekit-api = { workspace = true, features = ["rustls-tls-native-roots"] } +livekit-capture = { workspace = true } yuv-sys = { workspace = true, features = ["jpeg"] } futures = { workspace = true } clap = { workspace = true, features = ["derive"] } diff --git a/examples/local_video/src/encoded_publisher.rs b/examples/local_video/src/encoded_publisher.rs new file mode 100644 index 000000000..c230c5708 --- /dev/null +++ b/examples/local_video/src/encoded_publisher.rs @@ -0,0 +1,325 @@ +//! Publishes a pre-encoded H.264/H.265 stream received over TCP, without +//! re-encoding. +//! +//! The stream is length-prefixed NAL units (`avc3`/`hev1` framing, parameter +//! sets in-band), as produced by hardware encoder bridges. Access units are +//! assembled with `livekit-capture`'s `LengthPrefixedAccessUnitParser` and fed +//! to an `EncodedVideoPump`. + +use anyhow::Result; +use clap::{Parser, ValueEnum}; +use livekit::options::{FrameMetadataFeatures, TrackPublishOptions}; +use livekit::prelude::*; +use livekit_api::access_token; +use livekit_capture::{ + encoded::{ + h26x::{frame_metadata_from_sei, LengthPrefixedAccessUnitParser}, + EncodedVideoCodec, EncodedVideoPump, EncodedVideoSource, OwnedEncodedAccessUnit, + }, + error::SourceError, + primitive::VideoResolution, + pump::PumpStop, +}; +use log::info; +use std::env; +use std::io::{ErrorKind, Read}; +use std::net::TcpStream; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum Codec { + H264, + H265, +} + +#[derive(Parser, Debug)] +#[command(author, version, about, long_about = None)] +struct Args { + /// TCP address of the encoder's length-prefixed stream + #[arg(long, default_value = "127.0.0.1:5004")] + src: String, + + /// Codec of the stream + #[arg(long, value_enum, default_value_t = Codec::H265)] + codec: Codec, + + /// Size of each NAL length prefix in bytes + #[arg(long, default_value_t = 4)] + nal_length_size: u8, + + /// Publish each picture as soon as its slice arrives (single-slice streams) + #[arg(long, default_value_t = false)] + single_slice_flush: bool, + + /// Attach the user timestamp and frame id from an LKTS SEI as packet trailers + #[arg(long, default_value_t = false)] + packet_trailer: bool, + + /// Stream width + #[arg(long, default_value_t = 1920)] + width: u32, + + /// Stream height + #[arg(long, default_value_t = 1080)] + height: u32, + + /// Log the sender's pacer delay and bandwidth estimate every 2 s + #[arg(long, default_value_t = false)] + stats: bool, + + /// LiveKit participant identity + #[arg(long, default_value = "encoded-publisher")] + identity: String, + + /// LiveKit room name + #[arg(long, default_value = "video-room")] + room_name: String, + + /// LiveKit server URL + #[arg(long)] + url: Option, + + /// LiveKit API key (can also be set via LIVEKIT_API_KEY environment variable) + #[arg(long)] + api_key: Option, + + /// LiveKit API secret (can also be set via LIVEKIT_API_SECRET environment variable) + #[arg(long)] + api_secret: Option, +} + +/// Reads length-prefixed access units from a TCP stream. +/// +/// The connection is opened on the first read, i.e. once the pump runs +/// after the track is published, so no frames queue up in the socket while +/// the publication is negotiated. +struct TcpLengthPrefixedSource { + addr: String, + stream: Option, + parser: LengthPrefixedAccessUnitParser, + codec: EncodedVideoCodec, + resolution: VideoResolution, + buf: Vec, + eof: bool, + keyframe_requests: u64, +} + +impl TcpLengthPrefixedSource { + fn stream(&mut self) -> Result<&mut TcpStream, SourceError> { + if self.stream.is_none() { + let stream = TcpStream::connect(&self.addr).map_err(SourceError::new)?; + stream.set_nodelay(true).map_err(SourceError::new)?; + // Bounded reads so a stop request is noticed promptly. + stream.set_read_timeout(Some(Duration::from_millis(100))).map_err(SourceError::new)?; + info!("Connected to encoder at {}", self.addr); + self.stream = Some(stream); + } + Ok(self.stream.as_mut().unwrap()) + } +} + +fn now_us() -> i64 { + SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64 +} + +impl EncodedVideoSource for TcpLengthPrefixedSource { + fn resolution(&self) -> VideoResolution { + self.resolution + } + + fn codec(&self) -> EncodedVideoCodec { + self.codec + } + + fn next_access_unit( + &mut self, + stop: &PumpStop, + ) -> Result, SourceError> { + // Drain access units already buffered before reading more. + let mut bytes_read = 0; + loop { + let unit = if self.eof { + self.parser.flush() + } else { + self.parser.push(&self.buf[..bytes_read]) + } + .map_err(SourceError::new)?; + if let Some(mut unit) = unit { + // Stamp the capture time on arrival rather than the + // parser's nominal frame clock. + unit.timestamp_us = now_us(); + return Ok(Some(unit)); + } + if self.eof || stop.is_stopped() { + return Ok(None); + } + + let mut buf = std::mem::take(&mut self.buf); + let read = self.stream()?.read(&mut buf); + self.buf = buf; + bytes_read = match read { + Ok(0) => { + self.eof = true; + 0 + } + Ok(n) => n, + Err(err) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => 0, + Err(err) if err.kind() == ErrorKind::Interrupted => 0, + Err(err) => return Err(SourceError::new(err)), + }; + } + } + + fn request_keyframe(&mut self) { + // A real encoder bridge would ask the encoder for an IDR here. + self.keyframe_requests += 1; + log::debug!("keyframe requested ({} so far)", self.keyframe_requests); + } +} + +#[tokio::main] +async fn main() -> Result<()> { + env_logger::init(); + let args = Args::parse(); + + let url = args + .url + .or_else(|| env::var("LIVEKIT_URL").ok()) + .expect("LIVEKIT_URL must be provided via --url or env"); + let api_key = args + .api_key + .or_else(|| env::var("LIVEKIT_API_KEY").ok()) + .expect("LIVEKIT_API_KEY must be provided via --api-key or env"); + let api_secret = args + .api_secret + .or_else(|| env::var("LIVEKIT_API_SECRET").ok()) + .expect("LIVEKIT_API_SECRET must be provided via --api-secret or env"); + + let token = access_token::AccessToken::with_api_key(&api_key, &api_secret) + .with_identity(&args.identity) + .with_name(&args.identity) + .with_grants(access_token::VideoGrants { + room_join: true, + room: args.room_name.clone(), + can_publish: Some(true), + can_subscribe: Some(false), + ..Default::default() + }) + .to_jwt()?; + + info!("Connecting to LiveKit room '{}' as '{}'...", args.room_name, args.identity); + let (room, _events) = Room::connect(&url, &token, RoomOptions::default()).await?; + + let codec = match args.codec { + Codec::H264 => EncodedVideoCodec::H264, + Codec::H265 => EncodedVideoCodec::H265, + }; + let resolution = VideoResolution::new(args.width, args.height); + let parser = + LengthPrefixedAccessUnitParser::new(codec, args.nal_length_size, 0, 0, resolution)? + .with_single_slice_flush(args.single_slice_flush); + let source = TcpLengthPrefixedSource { + addr: args.src.clone(), + stream: None, + parser, + codec, + resolution, + buf: vec![0; 256 * 1024], + eof: false, + keyframe_requests: 0, + }; + + let mut pump = EncodedVideoPump::new(source); + if args.packet_trailer { + pump = pump.with_frame_metadata(move |unit| frame_metadata_from_sei(codec, &unit.payload)); + } + + let track = LocalVideoTrack::create_video_track("encoded", pump.rtc_source()); + let mut frame_metadata_features = FrameMetadataFeatures::default(); + frame_metadata_features.user_timestamp = args.packet_trailer; + frame_metadata_features.frame_id = args.packet_trailer; + let publish_options = TrackPublishOptions { + source: TrackSource::Camera, + frame_metadata_features, + ..pump.publish_options() + }; + room.local_participant() + .publish_track(LocalTrack::Video(track.clone()), publish_options) + .await?; + info!( + "Published {:?} track from {} (single-slice flush: {}, packet trailer: {})", + args.codec, args.src, args.single_slice_flush, args.packet_trailer + ); + + if args.stats { + tokio::spawn(log_sender_stats(track)); + } + + let running = pump.spawn()?; + let stop = running.stop_handle(); + let join = running.join(); + tokio::pin!(join); + let stats = tokio::select! { + stats = &mut join => stats, + _ = tokio::signal::ctrl_c() => { + info!("Ctrl-C received, exiting..."); + stop.stop(); + join.await + } + }?; + info!("Encoder stream ended: {:?}", stats); + room.close().await?; + Ok(()) +} + +/// Logs, from standard WebRTC stats, the negotiated codec and frames sent, +/// how long packets wait in the pacer (`totalPacketSendDelay` / +/// `packetsSent` over the interval), the sender's target bitrate and the +/// bandwidth estimate. +async fn log_sender_stats(track: LocalVideoTrack) { + use livekit::webrtc::stats::RtcStats; + + let mut interval = tokio::time::interval(Duration::from_secs(2)); + let (mut last_delay, mut last_packets) = (0.0, 0); + loop { + interval.tick().await; + let Ok(stats) = track.get_stats().await else { continue }; + let mut available_outgoing_bitrate = 0.0; + let mut outbound = None; + let mut codecs = Vec::new(); + for stat in stats { + match stat { + RtcStats::OutboundRtp(rtp) if rtp.stream.kind == "video" => outbound = Some(rtp), + RtcStats::Codec(codec) => codecs.push(codec), + RtcStats::CandidatePair(pair) if pair.candidate_pair.nominated => { + available_outgoing_bitrate = pair.candidate_pair.available_outgoing_bitrate; + } + _ => {} + } + } + let Some(rtp) = outbound else { continue }; + let codec = codecs + .iter() + .find(|codec| codec.rtc.id == rtp.stream.codec_id) + .map_or("unknown", |codec| codec.codec.mime_type.as_str()); + let delay = rtp.outbound.total_packet_send_delay; + let packets = rtp.sent.packets_sent; + if packets > last_packets { + info!( + "{codec}, {} frames sent, pacer delay {:.2} ms/packet over {} packets, \ + target bitrate {:.0} kbps, available outgoing bitrate {:.0} kbps", + rtp.outbound.frames_sent, + (delay - last_delay) * 1000.0 / (packets - last_packets) as f64, + packets - last_packets, + rtp.outbound.target_bitrate / 1000.0, + available_outgoing_bitrate / 1000.0, + ); + } else { + info!( + "{codec}, {} frames sent, no packets sent in the last 2 s", + rtp.outbound.frames_sent + ); + } + (last_delay, last_packets) = (delay, packets); + } +} diff --git a/livekit-capture/src/encoded/h26x.rs b/livekit-capture/src/encoded/h26x.rs index 2f3e49983..8e1b1d736 100644 --- a/livekit-capture/src/encoded/h26x.rs +++ b/livekit-capture/src/encoded/h26x.rs @@ -20,6 +20,7 @@ use crate::{ primitive::VideoResolution, }; use bytes::Bytes; +use livekit::webrtc::video_frame::FrameMetadata; use std::ops::Range; use thiserror::Error; @@ -84,16 +85,24 @@ pub struct AnnexBAccessUnitParser { resolution: VideoResolution, } -/// H.264/AVC length-prefixed parser state. -#[cfg(test)] +/// Incremental parser that splits an H.264 (AVC) or H.265 (HVCC) stream of +/// length-prefixed NAL units into access units. +/// +/// This is the framing of `stream-format=avc3` (H.264) and `hev1` (H.265), +/// with parameter sets in-band. Each access unit is converted to Annex-B. +/// +/// Call [`LengthPrefixedAccessUnitParser::push`] as bytes arrive, and +/// [`LengthPrefixedAccessUnitParser::flush`] at the end of the stream. #[derive(Debug, Clone)] -pub(crate) struct AvcAccessUnitParser { +pub struct LengthPrefixedAccessUnitParser { + codec: EncodedVideoCodec, pending: Vec, /// Complete NAL ranges found in `pending`. nal_ranges: Vec>, /// Offset of the first unparsed length prefix or incomplete NAL in `pending`. scan_cursor: usize, nal_length_size: u8, + single_slice_flush: bool, next_timestamp_us: i64, frame_interval_us: i64, resolution: VideoResolution, @@ -234,39 +243,67 @@ impl AccessUnitParser for AnnexBAccessUnitParser { } } -#[cfg(test)] -impl AvcAccessUnitParser { - /// Creates a parser for H.264/AVC length-prefixed byte streams. - pub(crate) fn new( +impl LengthPrefixedAccessUnitParser { + /// Creates a parser for an H.264 or H.265 length-prefixed byte stream. + /// + /// `nal_length_size` is the size of each big-endian NAL length prefix, + /// 1 to 4 bytes (`lengthSizeMinusOne + 1` from `avcC`/`hvcC`; 4 for most + /// encoders). Access units get timestamps that start at + /// `start_timestamp_us` and step by `frame_interval_us`. + pub fn new( + codec: EncodedVideoCodec, nal_length_size: u8, start_timestamp_us: i64, frame_interval_us: i64, resolution: VideoResolution, ) -> Result { - validate_avc_nal_length_size(nal_length_size)?; + match codec { + EncodedVideoCodec::H264 | EncodedVideoCodec::H265 => {} + EncodedVideoCodec::VP8 | EncodedVideoCodec::VP9 | EncodedVideoCodec::AV1 => { + return Err(H26xParseError::UnsupportedCodec(codec)); + } + } + validate_nal_length_size(nal_length_size)?; Ok(Self { + codec, pending: Vec::new(), nal_ranges: Vec::new(), scan_cursor: 0, nal_length_size, + single_slice_flush: false, next_timestamp_us: start_timestamp_us, frame_interval_us, resolution, }) } - /// Pushes encoded bytes and returns the next complete access unit if one is found. - pub(crate) fn push( - &mut self, - bytes: &[u8], - ) -> Result, H26xParseError> { + /// Completes each access unit as soon as its first slice arrives, + /// instead of when the next access unit begins. + /// + /// The end of an access unit is otherwise only known from the NAL that + /// starts the next one, which holds every picture for a full frame + /// interval unless the encoder terminates access units with an AUD. + /// Enable this only for streams with one slice per picture, which is + /// what hardware encoders typically emit: with several slices per + /// picture, the remaining slices would land in the next access unit. + pub fn with_single_slice_flush(mut self, enabled: bool) -> Self { + self.single_slice_flush = enabled; + self + } + + /// Pushes encoded bytes and returns the next complete access unit, if + /// one is found. + /// + /// One call returns at most one access unit. Push an empty slice to + /// pull further access units that are already buffered. + pub fn push(&mut self, bytes: &[u8]) -> Result, H26xParseError> { self.pending.extend_from_slice(bytes); self.drain_next(false) } - /// Flushes the pending bytes as the final access unit. - pub(crate) fn flush(&mut self) -> Result, H26xParseError> { + /// Flushes the remaining buffered bytes as the final access unit. + pub fn flush(&mut self) -> Result, H26xParseError> { self.drain_next(true) } @@ -276,11 +313,7 @@ impl AvcAccessUnitParser { ) -> Result, H26xParseError> { self.scan_pending(at_eof)?; - if let Some(split_at) = avc_access_unit_split_index( - &self.pending, - &self.nal_ranges, - self.nal_length_size as usize, - )? { + if let Some(split_at) = self.split_index()? { return self.take_access_unit(split_at); } if at_eof && !self.nal_ranges.is_empty() { @@ -294,29 +327,54 @@ impl AvcAccessUnitParser { Ok(None) } + /// Returns the byte offset that ends the current access unit, if known. + fn split_index(&self) -> Result, H26xParseError> { + let nal_length_size = self.nal_length_size as usize; + if self.single_slice_flush { + // Scanned ranges only cover complete NALs, so the first VCL NAL + // already holds the whole picture. + for range in &self.nal_ranges { + if is_vcl_nal(self.codec, &self.pending[range.clone()])? { + return Ok(Some(range.end)); + } + } + return Ok(None); + } + match access_unit_boundary_nal(self.codec, &self.pending, &self.nal_ranges)? { + Some(index) => self.nal_ranges[index] + .start + .checked_sub(nal_length_size) + .ok_or(H26xParseError::InvalidEncodedData("missing NAL length prefix")) + .map(Some), + None => Ok(None), + } + } + /// Parses length-prefixed NAL units appended since the previous call. fn scan_pending(&mut self, at_eof: bool) -> Result<(), H26xParseError> { let nal_length_size = self.nal_length_size as usize; while self.scan_cursor < self.pending.len() { if self.pending.len() - self.scan_cursor < nal_length_size { if at_eof { - return Err(H26xParseError::InvalidEncodedData("truncated AVC NAL length")); + return Err(H26xParseError::InvalidEncodedData("truncated NAL length prefix")); } break; } let nal_start = self.scan_cursor + nal_length_size; - let nal_len = read_avc_nal_length(&self.pending[self.scan_cursor..nal_start]); + let nal_len = read_nal_length(&self.pending[self.scan_cursor..nal_start]); if nal_len == 0 { - return Err(H26xParseError::InvalidEncodedData("empty AVC NAL unit")); + return Err(H26xParseError::InvalidEncodedData("empty length-prefixed NAL unit")); } let Some(nal_end) = nal_start.checked_add(nal_len) else { - return Err(H26xParseError::InvalidEncodedData("AVC NAL unit length overflow")); + return Err(H26xParseError::InvalidEncodedData("NAL unit length overflow")); }; if nal_end > self.pending.len() { if at_eof { - return Err(H26xParseError::InvalidEncodedData("truncated AVC NAL unit")); + return Err(H26xParseError::InvalidEncodedData( + "truncated length-prefixed NAL unit", + )); } break; } @@ -335,32 +393,56 @@ impl AvcAccessUnitParser { return Ok(None); } - let access_unit = self.pending[..byte_len].to_vec(); - self.pending.drain(..byte_len); - self.nal_ranges.retain_mut(|range| { - if range.end <= byte_len { - return false; + // The NAL ranges are already known: build the Annex-B payload and + // classify the frame from them instead of re-parsing the bytes. + let nal_count = self.nal_ranges.iter().take_while(|range| range.end <= byte_len).count(); + let nals = &self.nal_ranges[..nal_count]; + let nal_units: Vec<&[u8]> = nals.iter().map(|range| &self.pending[range.clone()]).collect(); + let frame_type = if is_keyframe_nalus(self.codec, &nal_units)? { + EncodedFrameType::Key + } else { + EncodedFrameType::Delta + }; + let payload = if self.nal_length_size as usize == ANNEX_B_START_CODE.len() { + // A 4-byte prefix is the size of a start code: convert in place + // during the one copy out of the pending buffer. + let mut payload = self.pending[..byte_len].to_vec(); + for range in nals { + payload[range.start - ANNEX_B_START_CODE.len()..range.start] + .copy_from_slice(&ANNEX_B_START_CODE); } + payload + } else { + annex_b_payload(&nal_units)? + }; + + self.pending.drain(..byte_len); + self.nal_ranges.drain(..nal_count); + for range in &mut self.nal_ranges { range.start -= byte_len; range.end -= byte_len; - true - }); + } self.scan_cursor -= byte_len; let timestamp_us = self.next_timestamp_us; self.next_timestamp_us = self.next_timestamp_us.saturating_add(self.frame_interval_us); - access_unit_from_h264_avc(&access_unit, self.nal_length_size, timestamp_us, self.resolution) - .map(Some) + Ok(Some(OwnedEncodedAccessUnit::new( + self.codec, + payload, + timestamp_us, + frame_type, + self.resolution, + ))) } } #[cfg(test)] -impl AccessUnitParser for AvcAccessUnitParser { +impl AccessUnitParser for LengthPrefixedAccessUnitParser { fn push(&mut self, bytes: &[u8]) -> Result, H26xParseError> { - AvcAccessUnitParser::push(self, bytes) + LengthPrefixedAccessUnitParser::push(self, bytes) } fn flush(&mut self) -> Result, H26xParseError> { - AvcAccessUnitParser::flush(self) + LengthPrefixedAccessUnitParser::flush(self) } } @@ -406,8 +488,40 @@ pub fn access_unit_from_h264_avc( timestamp_us: i64, resolution: VideoResolution, ) -> Result { - let nals = avc_nalus(payload, nal_length_size)?; - access_unit_from_nalus(EncodedVideoCodec::H264, &nals, timestamp_us, resolution) + access_unit_from_length_prefixed( + EncodedVideoCodec::H264, + payload, + nal_length_size, + timestamp_us, + resolution, + ) +} + +/// Creates an Annex-B access unit from H.265/HVCC length-prefixed NAL units. +pub fn access_unit_from_h265_hvcc( + payload: &[u8], + nal_length_size: u8, + timestamp_us: i64, + resolution: VideoResolution, +) -> Result { + access_unit_from_length_prefixed( + EncodedVideoCodec::H265, + payload, + nal_length_size, + timestamp_us, + resolution, + ) +} + +fn access_unit_from_length_prefixed( + codec: EncodedVideoCodec, + payload: &[u8], + nal_length_size: u8, + timestamp_us: i64, + resolution: VideoResolution, +) -> Result { + let nals = length_prefixed_nalus(payload, nal_length_size)?; + access_unit_from_nalus(codec, &nals, timestamp_us, resolution) } /// Creates an access unit from an Annex-B buffer. @@ -481,6 +595,143 @@ fn is_keyframe_nalus( } } +/// UUID of the `user_data_unregistered` SEI that carries an LKTS packet +/// trailer, as written by the LiveKit Go SDK and its ingest tooling. +const PACKET_TRAILER_SEI_UUID: [u8; 16] = [ + 0x3f, 0xa8, 0x5f, 0x64, 0x57, 0x17, 0x45, 0x62, 0xb3, 0xfc, 0x2c, 0x96, 0x3f, 0x66, 0xaf, 0xa6, +]; +const PACKET_TRAILER_MAGIC: &[u8; 4] = b"LKTS"; +const PACKET_TRAILER_TAG_USER_TIMESTAMP: u8 = 0x01; +const PACKET_TRAILER_TAG_FRAME_ID: u8 = 0x02; + +/// Returns the packet-trailer metadata embedded in an Annex-B access unit's +/// LKTS SEI, if it carries one. +/// +/// Encoders that know a frame's capture time can prefix the access unit with +/// a `user_data_unregistered` SEI holding an LKTS packet trailer (user +/// timestamp and frame id), the convention the LiveKit Go SDK reads with +/// `ReaderTrackWithPacketTrailer`. Pass the result to +/// [`EncodedVideoPump::with_frame_metadata`](crate::encoded::EncodedVideoPump::with_frame_metadata) +/// to publish it as the frame's packet trailer. The SEI itself is left in +/// the bitstream; decoders ignore unregistered user data. +/// +/// Only NAL units ahead of the first slice are inspected. +pub fn frame_metadata_from_sei(codec: EncodedVideoCodec, annex_b: &[u8]) -> Option { + let (sei_type, header_len) = match codec { + EncodedVideoCodec::H264 => (6, 1), + EncodedVideoCodec::H265 => (39, 2), + EncodedVideoCodec::VP8 | EncodedVideoCodec::VP9 | EncodedVideoCodec::AV1 => return None, + }; + let (offset, prefix_len) = find_start_code(annex_b)?; + let mut nal_start = offset + prefix_len; + loop { + let next = find_start_code(&annex_b[nal_start..]); + let nal_end = next.map_or(annex_b.len(), |(offset, _)| nal_start + offset); + let nal = &annex_b[nal_start..nal_end]; + if nal.len() > header_len { + let nal_type = match codec { + EncodedVideoCodec::H264 => nal[0] & 0x1f, + _ => (nal[0] >> 1) & 0x3f, + }; + if is_vcl_nal(codec, nal).unwrap_or(true) { + return None; + } + if nal_type == sei_type { + if let Some(metadata) = frame_metadata_from_sei_rbsp(&nal[header_len..]) { + return Some(metadata); + } + } + } + let (offset, prefix_len) = next?; + nal_start += offset + prefix_len; + } +} + +/// Parses the SEI messages of one SEI NAL (after its header) for an LKTS +/// `user_data_unregistered` payload. +fn frame_metadata_from_sei_rbsp(ebsp: &[u8]) -> Option { + let rbsp = remove_emulation_prevention(ebsp); + let mut i = 0; + // more_rbsp_data(): stop at the trailing stop bit. + while i + 1 < rbsp.len() { + let mut payload_type = 0usize; + while *rbsp.get(i)? == 0xff { + payload_type += 255; + i += 1; + } + payload_type += usize::from(*rbsp.get(i)?); + i += 1; + let mut payload_size = 0usize; + while *rbsp.get(i)? == 0xff { + payload_size += 255; + i += 1; + } + payload_size += usize::from(*rbsp.get(i)?); + i += 1; + let payload = rbsp.get(i..i.checked_add(payload_size)?)?; + i += payload_size; + if payload_type == 5 && payload.starts_with(&PACKET_TRAILER_SEI_UUID) { + return parse_packet_trailer(&payload[PACKET_TRAILER_SEI_UUID.len()..]); + } + } + None +} + +/// Parses an LKTS packet trailer: XOR-masked TLVs, then the masked trailer +/// length and the `LKTS` magic. +fn parse_packet_trailer(data: &[u8]) -> Option { + let magic_start = data.len().checked_sub(PACKET_TRAILER_MAGIC.len())?; + if &data[magic_start..] != PACKET_TRAILER_MAGIC { + return None; + } + let trailer_len = usize::from(*data.get(magic_start.checked_sub(1)?)? ^ 0xff); + // Envelope: length byte plus magic. + let envelope = 1 + PACKET_TRAILER_MAGIC.len(); + if trailer_len < envelope || trailer_len > data.len() { + return None; + } + let tlvs = &data[data.len() - trailer_len..magic_start - 1]; + + let mut metadata = FrameMetadata { user_timestamp: None, frame_id: None, user_data: None }; + let mut found = false; + let mut pos = 0; + while pos + 2 <= tlvs.len() { + let tag = tlvs[pos] ^ 0xff; + let len = usize::from(tlvs[pos + 1] ^ 0xff); + pos += 2; + let Some(value) = tlvs.get(pos..pos + len) else { break }; + let unmasked = || value.iter().fold(0u64, |acc, byte| (acc << 8) | u64::from(byte ^ 0xff)); + match (tag, len) { + (PACKET_TRAILER_TAG_USER_TIMESTAMP, 8) => { + metadata.user_timestamp = Some(unmasked()); + found = true; + } + (PACKET_TRAILER_TAG_FRAME_ID, 4) => { + metadata.frame_id = Some(unmasked() as u32); + found = true; + } + _ => {} + } + pos += len; + } + found.then_some(metadata) +} + +/// Strips `emulation_prevention_three_byte`s (`00 00 03` -> `00 00`). +fn remove_emulation_prevention(ebsp: &[u8]) -> Vec { + let mut rbsp = Vec::with_capacity(ebsp.len()); + let mut zeros = 0; + for &byte in ebsp { + if zeros >= 2 && byte == 0x03 { + zeros = 0; + continue; + } + zeros = if byte == 0 { zeros + 1 } else { 0 }; + rbsp.push(byte); + } + rbsp +} + fn h264_nal_type(nal: &[u8]) -> Result { let header = nal.first().ok_or(H26xParseError::EmptyPayload)?; Ok(header & 0x1f) @@ -527,22 +778,6 @@ fn access_unit_split_index( } } -#[cfg(test)] -fn avc_access_unit_split_index( - bytes: &[u8], - ranges: &[Range], - nal_length_size: usize, -) -> Result, H26xParseError> { - match access_unit_boundary_nal(EncodedVideoCodec::H264, bytes, ranges)? { - Some(index) => ranges[index] - .start - .checked_sub(nal_length_size) - .ok_or(H26xParseError::InvalidEncodedData("missing AVC NAL length")) - .map(Some), - None => Ok(None), - } -} - /// Returns the index of the first NAL that starts a new access unit, once at /// least one VCL NAL has been seen in the current one. fn access_unit_boundary_nal( @@ -621,34 +856,39 @@ fn is_vcl_nal(codec: EncodedVideoCodec, nal: &[u8]) -> Result Option<(usize, usize)> { - let mut idx = 0; - while idx + 3 <= bytes.len() { - if bytes[idx..].starts_with(&[0, 0, 1]) { - return Some((idx, 3)); - } - if idx + 4 <= bytes.len() && bytes[idx..].starts_with(&[0, 0, 0, 1]) { - return Some((idx, 4)); + // Every start code ends in 0x01 after two zero bytes; jumping between + // 0x01 bytes and looking back is far cheaper than matching at every + // offset of a multi-kilobyte slice. + let mut from = 2; + while from < bytes.len() { + let one = from + bytes[from..].iter().position(|&byte| byte == 1)?; + if bytes[one - 1] == 0 && bytes[one - 2] == 0 { + return Some(if one >= 3 && bytes[one - 3] == 0 { (one - 3, 4) } else { (one - 2, 3) }); } - idx += 1; + from = one + 1; } None } -fn avc_nalus(payload: &[u8], nal_length_size: u8) -> Result, H26xParseError> { - let ranges = avc_nal_ranges(payload, nal_length_size, true)?; +fn length_prefixed_nalus( + payload: &[u8], + nal_length_size: u8, +) -> Result, H26xParseError> { + let ranges = length_prefixed_nal_ranges(payload, nal_length_size, true)?; if ranges.is_empty() { return Err(H26xParseError::EmptyPayload); } Ok(ranges.into_iter().map(|range| &payload[range]).collect()) } -fn avc_nal_ranges( +fn length_prefixed_nal_ranges( bytes: &[u8], nal_length_size: u8, at_eof: bool, ) -> Result>, H26xParseError> { - validate_avc_nal_length_size(nal_length_size)?; + validate_nal_length_size(nal_length_size)?; let nal_length_size = nal_length_size as usize; let mut ranges = Vec::new(); @@ -656,23 +896,25 @@ fn avc_nal_ranges( while cursor < bytes.len() { if bytes.len() - cursor < nal_length_size { if at_eof { - return Err(H26xParseError::InvalidEncodedData("truncated AVC NAL length")); + return Err(H26xParseError::InvalidEncodedData("truncated NAL length prefix")); } break; } - let nal_len = read_avc_nal_length(&bytes[cursor..cursor + nal_length_size]); + let nal_len = read_nal_length(&bytes[cursor..cursor + nal_length_size]); cursor += nal_length_size; if nal_len == 0 { - return Err(H26xParseError::InvalidEncodedData("empty AVC NAL unit")); + return Err(H26xParseError::InvalidEncodedData("empty length-prefixed NAL unit")); } let Some(nal_end) = cursor.checked_add(nal_len) else { - return Err(H26xParseError::InvalidEncodedData("AVC NAL unit length overflow")); + return Err(H26xParseError::InvalidEncodedData("NAL unit length overflow")); }; if nal_end > bytes.len() { if at_eof { - return Err(H26xParseError::InvalidEncodedData("truncated AVC NAL unit")); + return Err(H26xParseError::InvalidEncodedData( + "truncated length-prefixed NAL unit", + )); } break; } @@ -684,15 +926,15 @@ fn avc_nal_ranges( Ok(ranges) } -fn read_avc_nal_length(bytes: &[u8]) -> usize { +fn read_nal_length(bytes: &[u8]) -> usize { bytes.iter().fold(0usize, |len, byte| (len << 8) | usize::from(*byte)) } -fn validate_avc_nal_length_size(nal_length_size: u8) -> Result<(), H26xParseError> { +fn validate_nal_length_size(nal_length_size: u8) -> Result<(), H26xParseError> { if (1..=4).contains(&nal_length_size) { return Ok(()); } - Err(H26xParseError::InvalidEncodedData("invalid AVC NAL length size")) + Err(H26xParseError::InvalidEncodedData("invalid NAL length prefix size")) } #[cfg(test)] @@ -780,7 +1022,207 @@ mod tests { access_unit_from_h264_avc(&[0, 0, 0, 3, 0x65], 4, 10, VideoResolution::new(640, 480)) .unwrap_err(); - assert_eq!(err, H26xParseError::InvalidEncodedData("truncated AVC NAL unit")); + assert_eq!(err, H26xParseError::InvalidEncodedData("truncated length-prefixed NAL unit")); + } + + #[test] + fn access_unit_from_hvcc_converts_length_prefixed_nals() { + let bytes = [ + 0, 0, 0, 3, 0x40, 0x01, 0x0c, // VPS + 0, 0, 0, 3, 0x42, 0x01, 0x02, // SPS + 0, 0, 0, 3, 0x44, 0x01, 0x03, // PPS + 0, 0, 0, 4, 0x26, 0x01, 0xaf, 0x08, // IDR_W_RADL + ]; + let au = access_unit_from_h265_hvcc(&bytes, 4, 10, VideoResolution::new(640, 480)).unwrap(); + + assert_eq!(au.codec, EncodedVideoCodec::H265); + assert_eq!(au.frame_type, EncodedFrameType::Key); + assert_eq!( + au.payload.as_ref(), + &[ + 0, 0, 0, 1, 0x40, 0x01, 0x0c, 0, 0, 0, 1, 0x42, 0x01, 0x02, 0, 0, 0, 1, 0x44, 0x01, + 0x03, 0, 0, 0, 1, 0x26, 0x01, 0xaf, 0x08, + ] + ); + } + + #[test] + fn access_unit_from_hvcc_marks_trailing_picture_as_delta() { + let au = access_unit_from_h265_hvcc( + &[0, 0, 0, 4, 0x02, 0x01, 0xd0, 0x09], + 4, + 10, + VideoResolution::new(640, 480), + ) + .unwrap(); + assert_eq!(au.frame_type, EncodedFrameType::Delta); + } + + /// An H.265 stream laid out like streaming-tools' gstreamer-source: + /// prefix SEI, parameter sets on keyframes, one slice, trailing AUD. + const H265_HEV1_STREAM: [u8; 71] = [ + 0, 0, 0, 3, 0x4e, 0x01, 0x05, // prefix SEI + 0, 0, 0, 3, 0x40, 0x01, 0x0c, // VPS + 0, 0, 0, 3, 0x42, 0x01, 0x02, // SPS + 0, 0, 0, 3, 0x44, 0x01, 0x03, // PPS + 0, 0, 0, 4, 0x26, 0x01, 0xaf, 0x08, // IDR_W_RADL + 0, 0, 0, 3, 0x46, 0x01, 0x50, // AUD + 0, 0, 0, 3, 0x4e, 0x01, 0x05, // prefix SEI + 0, 0, 0, 4, 0x02, 0x01, 0xd0, 0x09, // TRAIL_R + 0, 0, 0, 3, 0x46, 0x01, 0x50, // AUD + 0, 0, 0, 2, 0x4e, 0x01, // truncated SEI header (never completes) + ]; + + #[test] + fn hvcc_parser_waits_for_next_access_unit_without_single_slice_flush() { + let mut parser = LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::H265, + 4, + 0, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap(); + + // The IDR alone does not complete the access unit. + assert!(parser.push(&H265_HEV1_STREAM[..36]).unwrap().is_none()); + // The terminating AUD does: it starts the next access unit. + let au = parser.push(&H265_HEV1_STREAM[36..43]).unwrap().unwrap(); + assert_eq!(au.frame_type, EncodedFrameType::Key); + assert_eq!(annex_b_nalus(&au.payload).len(), 5); + } + + #[test] + fn hvcc_parser_single_slice_flush_completes_at_the_slice() { + let mut parser = LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::H265, + 4, + 0, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap() + .with_single_slice_flush(true); + + // Everything but the last byte of the IDR: nothing to publish yet. + assert!(parser.push(&H265_HEV1_STREAM[..35]).unwrap().is_none()); + // The last byte of the slice completes the picture immediately. + let au = parser.push(&H265_HEV1_STREAM[35..36]).unwrap().unwrap(); + assert_eq!(au.frame_type, EncodedFrameType::Key); + assert_eq!(au.timestamp_us, 0); + assert_eq!( + annex_b_nalus(&au.payload).iter().map(|nal| (nal[0] >> 1) & 0x3f).collect::>(), + vec![39, 32, 33, 34, 19] + ); + + // The trailing AUD leads the next access unit. + let au = parser.push(&H265_HEV1_STREAM[36..]).unwrap().unwrap(); + assert_eq!(au.frame_type, EncodedFrameType::Delta); + assert_eq!(au.timestamp_us, 33_333); + assert_eq!( + annex_b_nalus(&au.payload).iter().map(|nal| (nal[0] >> 1) & 0x3f).collect::>(), + vec![35, 39, 1] + ); + assert!(parser.drain().unwrap().is_none()); + } + + #[test] + fn length_prefixed_parser_rejects_unsupported_codec() { + let err = LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::VP8, + 4, + 0, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap_err(); + assert_eq!(err, H26xParseError::UnsupportedCodec(EncodedVideoCodec::VP8)); + } + + /// LKTS SEI as written by streaming-tools' gstreamer-source: user + /// timestamp 1_761_534_000_123_456, frame id 42. + const LKTS_SEI_H265: [u8; 42] = [ + 0x4e, 0x01, 0x05, 0x25, 0x3f, 0xa8, 0x5f, 0x64, 0x57, 0x17, 0x45, 0x62, 0xb3, 0xfc, 0x2c, + 0x96, 0x3f, 0x66, 0xaf, 0xa6, 0xfe, 0xf7, 0xff, 0xf9, 0xbd, 0xe4, 0xe7, 0xa9, 0xb1, 0xbf, + 0xfd, 0xfb, 0xff, 0xff, 0xff, 0xd5, 0xea, 0x4c, 0x4b, 0x54, 0x53, 0x80, + ]; + + fn annex_b(nals: &[&[u8]]) -> Vec { + annex_b_payload(nals).unwrap() + } + + #[test] + fn reads_frame_metadata_from_lkts_sei() { + let au = annex_b(&[&[0x46, 0x01, 0x50], &LKTS_SEI_H265, &[0x02, 0x01, 0xd0, 0x09]]); + let metadata = frame_metadata_from_sei(EncodedVideoCodec::H265, &au).unwrap(); + assert_eq!(metadata.user_timestamp, Some(1_761_534_000_123_456)); + assert_eq!(metadata.frame_id, Some(42)); + } + + #[test] + fn reads_frame_metadata_through_emulation_prevention() { + // The masked timestamp 0x0000_ffff_fc00_0000 contains 00 00 03. + let sei = [ + 0x4e, 0x01, 0x05, 0x25, 0x3f, 0xa8, 0x5f, 0x64, 0x57, 0x17, 0x45, 0x62, 0xb3, 0xfc, + 0x2c, 0x96, 0x3f, 0x66, 0xaf, 0xa6, 0xfe, 0xf7, 0xff, 0xff, 0x00, 0x00, 0x03, 0x03, + 0xff, 0xff, 0xff, 0xfd, 0xfb, 0xff, 0xff, 0xff, 0xf8, 0xea, 0x4c, 0x4b, 0x54, 0x53, + 0x80, + ]; + let au = annex_b(&[&sei, &[0x02, 0x01, 0xd0, 0x09]]); + let metadata = frame_metadata_from_sei(EncodedVideoCodec::H265, &au).unwrap(); + assert_eq!(metadata.user_timestamp, Some(0x0000_ffff_fc00_0000)); + assert_eq!(metadata.frame_id, Some(7)); + } + + #[test] + fn ignores_sei_after_the_first_slice_and_foreign_sei() { + let slice: &[u8] = &[0x02, 0x01, 0xd0, 0x09]; + let au = annex_b(&[slice, &LKTS_SEI_H265]); + assert!(frame_metadata_from_sei(EncodedVideoCodec::H265, &au).is_none()); + + // A user_data_unregistered SEI with another UUID. + let mut foreign = LKTS_SEI_H265; + foreign[4] ^= 0x01; + let au = annex_b(&[&foreign, slice]); + assert!(frame_metadata_from_sei(EncodedVideoCodec::H265, &au).is_none()); + } + + #[test] + fn reads_frame_metadata_from_h264_lkts_sei() { + // Same payload behind the one-byte H.264 SEI header. + let mut sei = vec![0x06]; + sei.extend_from_slice(&LKTS_SEI_H265[2..]); + let au = annex_b(&[&sei, &[0x41, 0x9a, 0x02]]); + let metadata = frame_metadata_from_sei(EncodedVideoCodec::H264, &au).unwrap(); + assert_eq!(metadata.frame_id, Some(42)); + } + + #[test] + fn find_start_code_matches_a_scan_at_every_offset() { + fn reference(bytes: &[u8]) -> Option<(usize, usize)> { + (0..bytes.len()).find_map(|idx| { + if bytes[idx..].starts_with(&[0, 0, 1]) { + Some((idx, 3)) + } else if bytes[idx..].starts_with(&[0, 0, 0, 1]) { + Some((idx, 4)) + } else { + None + } + }) + } + // Every string of up to 8 bytes over {0, 1, 2}. + for len in 0..=8u32 { + for mut n in 0..3usize.pow(len) { + let bytes: Vec = (0..len) + .map(|_| { + let byte = (n % 3) as u8; + n /= 3; + byte + }) + .collect(); + assert_eq!(find_start_code(&bytes), reference(&bytes), "{bytes:?}"); + } + } } #[test] @@ -821,8 +1263,14 @@ mod tests { #[test] fn avc_parser_splits_at_next_access_unit_delimiter() { - let mut parser = - AvcAccessUnitParser::new(4, 100, 33_333, VideoResolution::new(640, 480)).unwrap(); + let mut parser = LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::H264, + 4, + 100, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap(); let stream = [ 0, 0, 0, 2, 0x09, 0x10, 0, 0, 0, 3, 0x65, 1, 2, 0, 0, 0, 2, 0x09, 0x10, 0, 0, 0, 2, 0x41, 3, @@ -1069,12 +1517,45 @@ mod tests { 0, 0, 0, 3, 0x41, 0x9a, 0x03, // P ]; assert_chunked_matches_one_shot( - || AvcAccessUnitParser::new(4, 0, 33_333, VideoResolution::new(640, 480)).unwrap(), + || { + LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::H264, + 4, + 0, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap() + }, &h264_avc, 3, ); } + #[test] + fn chunked_length_prefixed_h265_matches_one_shot_parsing() { + // Drop the truncated tail so the stream flushes cleanly. + let stream = &H265_HEV1_STREAM[..65]; + for single_slice_flush in [false, true] { + assert_chunked_matches_one_shot( + || { + LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::H265, + 4, + 0, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap() + .with_single_slice_flush(single_slice_flush) + }, + // Both modes end with the final AUD flushed on its own. + stream, + 3, + ); + } + } + #[test] fn rejects_pending_access_unit_over_size_cap() { let mut parser = AnnexBAccessUnitParser::new( @@ -1095,8 +1576,14 @@ mod tests { #[test] fn avc_rejects_pending_access_unit_over_size_cap() { - let mut parser = - AvcAccessUnitParser::new(4, 0, 33_333, VideoResolution::new(640, 480)).unwrap(); + let mut parser = LengthPrefixedAccessUnitParser::new( + EncodedVideoCodec::H264, + 4, + 0, + 33_333, + VideoResolution::new(640, 480), + ) + .unwrap(); let nal_len = (MAX_PENDING_ACCESS_UNIT_BYTES + 1) as u32; assert!(parser.push(&nal_len.to_be_bytes()).unwrap().is_none()); diff --git a/livekit-capture/src/sources/gstreamer.rs b/livekit-capture/src/sources/gstreamer.rs index 65cdab506..35abfef70 100644 --- a/livekit-capture/src/sources/gstreamer.rs +++ b/livekit-capture/src/sources/gstreamer.rs @@ -26,7 +26,10 @@ use thiserror::Error; use crate::{ encoded::{ - h26x::{access_unit_from_annex_b, access_unit_from_h264_avc, H26xParseError}, + h26x::{ + access_unit_from_annex_b, access_unit_from_h264_avc, access_unit_from_h265_hvcc, + H26xParseError, + }, EncodedFrameType, EncodedVideoCodec, EncodedVideoSource, OwnedEncodedAccessUnit, }, error::SourceError, @@ -47,6 +50,11 @@ enum GStreamerSampleFormat { }, /// H.265 Annex-B access units, usually from `h265parse` with byte-stream caps. H265AnnexB, + /// H.265 access units with HVCC length-prefixed NAL units. + H265Hvcc { + /// Length-prefix size in bytes. + nal_length_size: u8, + }, /// One already-delimited encoded access unit per appsink sample. AccessUnit { /// Codec carried by each appsink sample. @@ -61,6 +69,7 @@ impl GStreamerSampleFormat { Self::H264AnnexB => EncodedVideoCodec::H264, Self::H264Avc { .. } => EncodedVideoCodec::H264, Self::H265AnnexB => EncodedVideoCodec::H265, + Self::H265Hvcc { .. } => EncodedVideoCodec::H265, Self::AccessUnit { codec } => codec, } } @@ -825,6 +834,9 @@ fn access_unit_from_sample_payload( timestamp_us, resolution, ), + GStreamerSampleFormat::H265Hvcc { nal_length_size } => { + access_unit_from_h265_hvcc(payload, nal_length_size, timestamp_us, resolution) + } GStreamerSampleFormat::AccessUnit { codec } => { if payload.is_empty() { return Err(H26xParseError::EmptyPayload); @@ -1196,9 +1208,14 @@ fn sample_format_from_caps_structure( Err(()) => return Ok(None), }; match stream_format.as_deref() { + // As with H.264 `avc`, `hvc1` keeps the parameter sets in + // `codec_data` only; `hev1` carries them in-band. + Some("hev1") => Ok(Some(GStreamerSampleFormat::H265Hvcc { + nal_length_size: h265_hvcc_nal_length_size_from_caps(structure), + })), Some("byte-stream") | None => Ok(Some(GStreamerSampleFormat::H265AnnexB)), Some(stream_format) => Err(GStreamerPipelineError::UnsupportedCaps(format!( - "H.265 stream-format '{stream_format}'; expected byte-stream" + "H.265 stream-format '{stream_format}'; expected byte-stream or hev1" ))), } } @@ -1246,6 +1263,23 @@ fn h264_avc_nal_length_size_from_codec_data(codec_data: &[u8]) -> Option { (1..=4).contains(&length_size).then_some(length_size) } +fn h265_hvcc_nal_length_size_from_caps(structure: &gst::StructureRef) -> u8 { + let Ok(codec_data) = structure.get::("codec_data") else { + return 4; + }; + let Ok(codec_data) = codec_data.map_readable() else { + return 4; + }; + h265_hvcc_nal_length_size_from_codec_data(codec_data.as_ref()).unwrap_or(4) +} + +/// Reads the HVCC NAL length-prefix size from `hvcC` codec data +/// (`lengthSizeMinusOne`, the low two bits of byte 21). +fn h265_hvcc_nal_length_size_from_codec_data(codec_data: &[u8]) -> Option { + let length_size = (codec_data.get(21)? & 0x03) + 1; + (1..=4).contains(&length_size).then_some(length_size) +} + /// Selects the single unlinked source pad that advertises a supported /// encoded video codec. /// @@ -1340,6 +1374,36 @@ mod tests { assert_eq!(access_unit.payload.as_ref(), &[0, 0, 0, 1, 0x65, 1, 2]); } + #[test] + fn sample_payload_h265_hvcc_converts_to_annex_b_and_detects_keyframe() { + let access_unit = access_unit_from_sample_payload( + GStreamerSampleFormat::H265Hvcc { nal_length_size: 4 }, + &[ + 0, 0, 0, 3, 0x40, 0x01, 0x0c, // VPS + 0, 0, 0, 3, 0x42, 0x01, 0x02, // SPS + 0, 0, 0, 3, 0x44, 0x01, 0x03, // PPS + 0, 0, 0, 4, 0x26, 0x01, 0xaf, 0x08, // IDR_W_RADL + ], + 1_000, + EncodedFrameType::Delta, + VideoResolution::new(640, 480), + ) + .unwrap(); + + assert_eq!(access_unit.codec, EncodedVideoCodec::H265); + assert_eq!(access_unit.frame_type, EncodedFrameType::Key); + assert_eq!(&access_unit.payload[..4], &[0, 0, 0, 1]); + assert_eq!(access_unit.payload.len(), 29); + } + + #[test] + fn hvcc_length_size_comes_from_codec_data() { + let mut hvcc = [0u8; 23]; + hvcc[21] = 0xfc | 0x01; // lengthSizeMinusOne = 1 + assert_eq!(h265_hvcc_nal_length_size_from_codec_data(&hvcc), Some(2)); + assert_eq!(h265_hvcc_nal_length_size_from_codec_data(&hvcc[..21]), None); + } + #[test] fn sample_payload_access_unit_uses_buffer_delta_flag() { let access_unit = access_unit_from_sample_payload( diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index 2b8c38e08..ceb97a595 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/livekit/src/room/participant/local_participant.rs @@ -376,9 +376,17 @@ impl LocalParticipant { async fn publish_track_with_video_send_encodings( &self, track: LocalTrack, - options: TrackPublishOptions, + mut options: TrackPublishOptions, video_send_encodings: Option>, ) -> RoomResult { + if options.video_encoder == options::VideoEncoderBackend::PreEncoded && options.simulcast { + // The pass-through forwards one encoded stream; it has no lower + // layers to offer. Simulcast would announce layers it cannot fill + // and send copies of the full stream on their SSRCs. + log::debug!("simulcast is not supported for pre-encoded video; publishing one layer"); + options.simulcast = false; + } + let disable_red = self.local.encryption_type != EncryptionType::None || !options.red; let mut req = proto::AddTrackRequest { @@ -431,6 +439,15 @@ impl LocalParticipant { layers: req.layers.clone(), ..Default::default() }]; + } else { + // Single-layer tracks still announce their codec: the + // server prefers it in the answer, and without it falls + // back to its default (VP8) for this transceiver. + req.simulcast_codecs = vec![proto::SimulcastCodec { + codec: options.video_codec.as_str().to_string(), + cid: track.rtc_track().id(), + ..Default::default() + }]; } } LocalTrack::Audio(_audio_track) => { diff --git a/livekit/src/rtc_engine/peer_transport.rs b/livekit/src/rtc_engine/peer_transport.rs index 7c64cbe6c..1b79b748d 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -39,7 +39,7 @@ struct TransportInner { max_send_bitrate_bps: Option, // Whether the target above belongs to a screen share, which is exempt from the // start bitrate cap. - max_send_bitrate_is_screen_share: bool, + max_send_bitrate_uncapped_start: bool, // How long this peer connection took to set up, set once the initial connect succeeds. // `compute_start_bitrate_kbps` lowers the hint for slow connections. connection_setup_time: Option, @@ -84,7 +84,7 @@ impl PeerTransport { restarting_ice: false, single_pc_mode, max_send_bitrate_bps: None, - max_send_bitrate_is_screen_share: false, + max_send_bitrate_uncapped_start: false, connection_setup_time: None, start_bitrate_applied: false, pending_initial_offer: None, @@ -257,10 +257,10 @@ impl PeerTransport { inner.pending_initial_offer = None; } - pub async fn set_max_send_bitrate_bps(&self, bps: Option, is_screen_share: bool) { + pub async fn set_max_send_bitrate_bps(&self, bps: Option, uncapped_start: bool) { let mut inner = self.inner.lock().await; inner.max_send_bitrate_bps = bps; - inner.max_send_bitrate_is_screen_share = is_screen_share; + inner.max_send_bitrate_uncapped_start = uncapped_start; } /// Maximum x-google-start-bitrate (kbps). @@ -294,7 +294,8 @@ impl PeerTransport { /// multiplier is used for every codec because the target already reflects the codec's /// efficiency. Camera is capped at [`Self::MAX_START_BITRATE_KBPS`] so the estimator does /// not open too aggressively on a high-bitrate track; screen share is exempt, because its - /// content needs the bitrate immediately to stay legible. + /// content needs the bitrate immediately to stay legible, and so are pre-encoded tracks, + /// whose bitrate is set upstream and only queues in the pacer behind a low estimate. /// /// Connection setup time (signaling join plus ICE/DTLS) is the only network signal there /// is before the first video offer, since libwebrtc cannot probe the path until a video @@ -635,7 +636,7 @@ impl PeerTransport { if has_video && !inner.start_bitrate_applied { if let Some(start_kbps) = Self::compute_start_bitrate_kbps( inner.max_send_bitrate_bps, - inner.max_send_bitrate_is_screen_share, + inner.max_send_bitrate_uncapped_start, inner.connection_setup_time, ) { log::info!( diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index 83d5540cc..0ebc85f4e 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -2048,8 +2048,13 @@ impl SessionInner { let sum: u64 = encodings.iter().filter_map(|e| e.max_bitrate).sum(); (sum > 0).then_some(sum) }; - let is_screen_share = options.source == TrackSource::Screenshare; - self.publisher_pc.set_max_send_bitrate_bps(ultimate_bps, is_screen_share).await; + // Screen share needs its bitrate immediately to stay legible. A + // pre-encoded track is already encoded at its own bitrate, which + // the sender cannot lower, so a capped starting estimate only + // queues its packets in the pacer until the estimate catches up. + let uncapped_start = options.source == TrackSource::Screenshare + || options.video_encoder == libwebrtc::rtp_sender::VideoEncoderBackend::PreEncoded; + self.publisher_pc.set_max_send_bitrate_bps(ultimate_bps, uncapped_start).await; } let init = RtpTransceiverInit { diff --git a/webrtc-sys/src/peer_connection_factory.cpp b/webrtc-sys/src/peer_connection_factory.cpp index 2595f27ad..5cb5cc8e8 100644 --- a/webrtc-sys/src/peer_connection_factory.cpp +++ b/webrtc-sys/src/peer_connection_factory.cpp @@ -17,6 +17,10 @@ #include "livekit/peer_connection_factory.h" #include +#include +#include +#include +#include #include #include @@ -119,11 +123,66 @@ class CompositeFieldTrials final : public webrtc::FieldTrialsView { std::vector> views_; }; +// Field trials from the LIVEKIT_WEBRTC_FIELD_TRIALS environment variable, in +// libwebrtc's "Name/Value/Name2/Value2/" format (the same as Chrome's +// --force-fieldtrials). Lets deployments tune WebRTC internals the SDK does +// not expose, e.g. the pacer with "WebRTC-Video-Pacing/factor:4.0/". +constexpr char kFieldTrialsEnvVar[] = "LIVEKIT_WEBRTC_FIELD_TRIALS"; + +class EnvFieldTrials final : public webrtc::FieldTrialsView { + public: + explicit EnvFieldTrials(std::map trials) + : trials_(std::move(trials)) {} + + // Returns nullptr when the variable is unset, empty, or malformed. + static std::unique_ptr FromEnvironment() { + const char* value = std::getenv(kFieldTrialsEnvVar); + if (value == nullptr || *value == '\0') { + return nullptr; + } + std::map trials; + std::string_view rest(value); + while (!rest.empty()) { + const size_t name_end = rest.find('/'); + const size_t value_end = + name_end == std::string_view::npos ? name_end + : rest.find('/', name_end + 1); + if (name_end == 0 || value_end == std::string_view::npos) { + RTC_LOG(LS_ERROR) << "Ignoring malformed " << kFieldTrialsEnvVar + << ": " << value; + return nullptr; + } + trials[std::string(rest.substr(0, name_end))] = + std::string(rest.substr(name_end + 1, value_end - name_end - 1)); + rest.remove_prefix(value_end + 1); + } + return std::make_unique(std::move(trials)); + } + + std::string Lookup(absl::string_view key) const override { + auto it = trials_.find(std::string(key)); + return it == trials_.end() ? "" : it->second; + } + + std::unique_ptr CreateCopy() const override { + return std::make_unique(trials_); + } + + private: + std::map trials_; +}; + // zero_playout_delay and enable_warp are independent and may both be enabled; -// their field-trial views are composed into one. +// their field-trial views are composed into one. Trials from the environment +// come first, so they override the SDK's own. webrtc::Environment CreateEnvironment(bool zero_playout_delay, bool enable_warp) { std::vector> views; + if (auto env_trials = EnvFieldTrials::FromEnvironment()) { + RTC_LOG(LS_INFO) << "WebRTC field trials from " << kFieldTrialsEnvVar + << ": " << std::getenv(kFieldTrialsEnvVar); + views.push_back(std::move(env_trials)); + } if (zero_playout_delay) { views.push_back(std::make_unique()); }