diff --git a/.changeset/negotiated_data_channels.md b/.changeset/negotiated_data_channels.md new file mode 100644 index 000000000..7054793ad --- /dev/null +++ b/.changeset/negotiated_data_channels.md @@ -0,0 +1,9 @@ +--- +livekit: minor +livekit-signaling: minor +livekit-api: minor +livekit-capture: minor +livekit-ffi: minor +--- + +Create the well-known data channels as negotiated (fixed IDs) when the server supports it (protocol 18), saving the data channel open handshake round trip. Falls back to in-band channels against older servers. diff --git a/livekit-signaling/src/lib.rs b/livekit-signaling/src/lib.rs index 226ad1372..31b44130d 100644 --- a/livekit-signaling/src/lib.rs +++ b/livekit-signaling/src/lib.rs @@ -59,7 +59,10 @@ pub const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); pub const SIGNAL_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); const REGION_FETCH_TIMEOUT: Duration = Duration::from_secs(3); const VALIDATE_TIMEOUT: Duration = Duration::from_secs(3); -pub const PROTOCOL_VERSION: u32 = 17; +pub const PROTOCOL_VERSION: u32 = 18; +/// First protocol version where the well-known data channels are created as negotiated +/// (out-of-band, fixed IDs) by both the client and the server. +pub const NEGOTIATED_DATA_CHANNELS_PROTOCOL: u32 = 18; /// Capabilities the Rust SDK advertises to the SFU at connect time. /// diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index 81ae60647..064793823 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -28,7 +28,9 @@ use libwebrtc::{prelude::*, stats::RtcStats}; use livekit_datatrack::backend as dt; use livekit_protocol::{self as proto}; use livekit_rpc::api::{RpcError, RpcErrorCode}; -use livekit_signaling::{SignalClient, SignalEvent, SignalEvents}; +use livekit_signaling::{ + SignalClient, SignalEvent, SignalEvents, NEGOTIATED_DATA_CHANNELS_PROTOCOL, +}; use parking_lot::Mutex; use prost::Message; use proto::SignalTarget; @@ -81,6 +83,11 @@ pub const ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); pub const TRACK_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); pub const LOSSY_DC_LABEL: &str = "_lossy"; pub const RELIABLE_DC_LABEL: &str = "_reliable"; + +// Fixed SCTP stream IDs used when the data channels are created as negotiated; must match the server. +const NEGOTIATED_RELIABLE_DC_ID: i32 = 100; +const NEGOTIATED_LOSSY_DC_ID: i32 = 101; +const NEGOTIATED_DATA_TRACK_DC_ID: i32 = 102; pub const DATA_TRACK_DC_LABEL: &str = "_data_track"; pub const RELIABLE_RECEIVED_STATE_TTL: Duration = Duration::from_secs(30); pub const PUBLISHER_NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); @@ -543,7 +550,8 @@ impl RtcSession { true, ); - let dcs = Self::create_data_channels(&publisher_pc, &emitter)?; + // always try to open negotiated data channels and close them if we know the server doesn't support it after receiving the join response + let dcs = Self::create_data_channels(&publisher_pc, &emitter, true)?; Self::add_recv_media_sections(&publisher_pc.peer_connection(), 3, 3)?; @@ -593,9 +601,28 @@ impl RtcSession { let (dc_emitter, dc_events) = mpsc::unbounded_channel(); + let server_protocol = + join_response.server_info.as_ref().map(|info| info.protocol).unwrap_or_default(); + // Dual peer connection has the server open the channels, so there is nothing to save. + let use_negotiated_dcs = + single_pc_mode && server_protocol >= NEGOTIATED_DATA_CHANNELS_PROTOCOL as i32; + log::debug!( + "server protocol {}, negotiated data channels: {}", + server_protocol, + use_negotiated_dcs + ); + let sent_publisher_offer; let (mut publisher_pc, mut reliable_dc, mut lossy_dc, data_track_dc) = - if let Some((pub_pc, dcs)) = early_publisher_pc { + if let Some((pub_pc, mut dcs)) = early_publisher_pc { + if !use_negotiated_dcs { + // The SCTP section is already in the offer, so in-band channels can be + // opened on it without sdp renegotiation. + dcs.0.close(); + dcs.1.close(); + dcs.2.close(); + dcs = Self::create_data_channels(&pub_pc, &emitter, false)?; + } if single_pc_mode { pub_pc.peer_connection().set_configuration(rtc_config.clone())?; sent_publisher_offer = publisher_offer.is_some(); @@ -612,7 +639,7 @@ impl RtcSession { proto::SignalTarget::Publisher, single_pc_mode, ); - let dcs = Self::create_data_channels(&publisher_pc, &emitter)?; + let dcs = Self::create_data_channels(&publisher_pc, &emitter, use_negotiated_dcs)?; (publisher_pc, dcs.0, dcs.1, dcs.2) }; @@ -743,22 +770,37 @@ impl RtcSession { fn create_data_channels( publisher_pc: &PeerTransport, emitter: &mpsc::UnboundedSender, + negotiated: bool, ) -> EngineResult<(DataChannel, DataChannel, DataChannel)> { let reliable_dc = publisher_pc.peer_connection().create_data_channel( RELIABLE_DC_LABEL, - DataChannelInit { ordered: true, ..Default::default() }, + DataChannelInit { + ordered: true, + negotiated, + id: if negotiated { NEGOTIATED_RELIABLE_DC_ID } else { -1 }, + ..Default::default() + }, )?; - let lossy_options = - DataChannelInit { ordered: false, max_retransmits: Some(0), ..Default::default() }; + let lossy_options = DataChannelInit { + ordered: false, + max_retransmits: Some(0), + negotiated, + id: if negotiated { NEGOTIATED_LOSSY_DC_ID } else { -1 }, + ..Default::default() + }; let lossy_dc = publisher_pc .peer_connection() .create_data_channel(LOSSY_DC_LABEL, lossy_options.clone())?; - let data_track_dc = publisher_pc - .peer_connection() - .create_data_channel(DATA_TRACK_DC_LABEL, lossy_options)?; + let data_track_dc = publisher_pc.peer_connection().create_data_channel( + DATA_TRACK_DC_LABEL, + DataChannelInit { + id: if negotiated { NEGOTIATED_DATA_TRACK_DC_ID } else { -1 }, + ..lossy_options + }, + )?; handle_remote_dt_packets(&data_track_dc, emitter.downgrade()); Ok((reliable_dc, lossy_dc, data_track_dc))