diff --git a/src/change/direct.rs b/src/change/direct.rs index 9e95df33a..067f118bd 100644 --- a/src/change/direct.rs +++ b/src/change/direct.rs @@ -284,6 +284,13 @@ impl<'a> DirectApi<'a> { self.rtc.session.streams.stream_rx_by_midrid(midrid, true) } + /// Get sctp max-message-size + /// + /// This returns the maximum buffer size that channel.write will accept + pub fn sctp_max_message_size(&self) -> u32 { + self.rtc.sctp.remote_max_message_size() + } + /// Declare the intention to send data using the given SSRC. /// /// * The resend RTX is optional but necessary to do resends. str0m does not do diff --git a/src/lib.rs b/src/lib.rs index 27a7df070..bff7441b9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1544,7 +1544,11 @@ impl Rtc { /// /// See [`Rtc`] instance documentation for how this is expected to be used in a loop. pub fn poll_output(&mut self) -> Result { - let o = self.do_poll_output()?; + let o = loop { + if let Some(o) = self.do_poll_output()? { + break o; + } + }; match &o { Output::Event(e) => match e { @@ -1572,10 +1576,10 @@ impl Rtc { Ok(o) } - fn do_poll_output(&mut self) -> Result { + fn do_poll_output(&mut self) -> Result, RtcError> { if self.state == RtcState::Closed { self.last_timeout_reason = Reason::NotHappening; - return Ok(Output::Timeout(not_happening())); + return Ok(Some(Output::Timeout(not_happening()))); } while let Some(e) = self.ice.poll_event() { @@ -1584,7 +1588,7 @@ impl Rtc { // } IceAgentEvent::IceConnectionStateChange(v) => { - return Ok(Output::Event(Event::IceConnectionStateChange(v))); + return Ok(Some(Output::Event(Event::IceConnectionStateChange(v)))); } IceAgentEvent::DiscoveredRecv { proto, source } => { debug!("ICE remote address: {:?}/{:?}", Pii(source), proto); @@ -1678,7 +1682,7 @@ impl Rtc { } DtlsOutput::CloseNotify => { self.start_close()?; - return Ok(Output::Event(Event::Closed)); + return Ok(Some(Output::Event(Event::Closed))); } other => { return Err(RtcError::Dtls(DtlsError::Io(std::io::Error::other( @@ -1689,7 +1693,7 @@ impl Rtc { } if just_connected { - return Ok(Output::Event(Event::Connected)); + return Ok(Some(Output::Event(Event::Connected))); } while let Some(e) = self.sctp.poll() { @@ -1723,13 +1727,13 @@ impl Rtc { // Run again since this would feed the DTLS subsystem // to produce a packet now. - return self.do_poll_output(); + return Ok(None); } } SctpEvent::Open { id, label } => { self.chan.ensure_channel_id_for(id); let id = self.chan.channel_id_by_stream_id(id).unwrap(); - return Ok(Output::Event(Event::ChannelOpen(id, label))); + return Ok(Some(Output::Event(Event::ChannelOpen(id, label)))); } SctpEvent::Close { id } => { let Some(id) = self.chan.channel_id_by_stream_id(id) else { @@ -1737,11 +1741,11 @@ impl Rtc { continue; }; self.chan.remove_channel(id, self.last_now); - return Ok(Output::Event(Event::ChannelClose(id))); + return Ok(Some(Output::Event(Event::ChannelClose(id)))); } SctpEvent::AssociationLost => { self.start_close()?; - return Ok(Output::Event(Event::Closed)); + return Ok(Some(Output::Event(Event::Closed))); } SctpEvent::Data { id, binary, data } => { let Some(id) = self.chan.channel_id_by_stream_id(id) else { @@ -1749,37 +1753,37 @@ impl Rtc { continue; }; let cd = ChannelData { id, binary, data }; - return Ok(Output::Event(Event::ChannelData(cd))); + return Ok(Some(Output::Event(Event::ChannelData(cd)))); } SctpEvent::BufferedAmountLow { id } => { let Some(id) = self.chan.channel_id_by_stream_id(id) else { warn!("Drop BufferedAmountLow for id: {:?}", id); continue; }; - return Ok(Output::Event(Event::ChannelBufferedAmountLow(id))); + return Ok(Some(Output::Event(Event::ChannelBufferedAmountLow(id)))); } } } if let Some(ev) = self.session.poll_event() { - return Ok(Output::Event(ev)); + return Ok(Some(Output::Event(ev))); } // Some polling needs to bubble up errors. if let Some(ev) = self.session.poll_event_fallible()? { - return Ok(Output::Event(ev)); + return Ok(Some(Output::Event(ev))); } if let Some(e) = self.stats.as_mut().and_then(|s| s.poll_output()) { - return Ok(match e { + return Ok(Some(match e { StatsEvent::Peer(s) => Output::Event(Event::PeerStats(s)), StatsEvent::MediaIngress(s) => Output::Event(Event::MediaIngressStats(s)), StatsEvent::MediaEgress(s) => Output::Event(Event::MediaEgressStats(s)), - }); + })); } if let Some(v) = self.ice.poll_transmit() { - return Ok(Output::Transmit(v)); + return Ok(Some(Output::Transmit(v))); } if let Some(send) = &self.send_addr { @@ -1795,7 +1799,7 @@ impl Rtc { destination: send.destination, contents, }; - return Ok(Output::Transmit(t)); + return Ok(Some(Output::Transmit(t))); } } else { // Don't allow accumulated feedback to build up indefinitely @@ -1827,11 +1831,11 @@ impl Rtc { if self.state == RtcState::Closing && self.close_drain_complete() { self.state = RtcState::Closed; self.last_timeout_reason = Reason::NotHappening; - return Ok(Output::Timeout(not_happening())); + return Ok(Some(Output::Timeout(not_happening()))); } self.last_timeout_reason = reason; - Ok(Output::Timeout(next)) + Ok(Some(Output::Timeout(next))) } /// The reason for the last [`Output::Timeout`] diff --git a/src/sctp/mod.rs b/src/sctp/mod.rs index 1ce119564..d0b70aadc 100644 --- a/src/sctp/mod.rs +++ b/src/sctp/mod.rs @@ -27,7 +27,7 @@ mod error; pub use error::SctpError; /// Bytes that can be buffered inside str0m across all streams. -const MAX_BUFFERED_ACROSS_STREAMS: usize = 128 * 1024; +const DEFAULT_MAX_BUFFERED_ACROSS_STREAMS: usize = 128 * 1024; /// Maximum message size we advertise in SDP (what we can receive) pub const LOCAL_MAX_MESSAGE_SIZE: u32 = 256 * 1024; @@ -35,6 +35,9 @@ pub const LOCAL_MAX_MESSAGE_SIZE: u32 = 256 * 1024; /// Default max message size if remote doesn't advertise pub const DEFAULT_REMOTE_MAX_MESSAGE_SIZE: u32 = 64 * 1024; +/// Headroom for max_buffered_across_streams vs remote_max_message_size +pub const BUFFER_MULTIPLIER: u32 = 2; + pub(crate) struct RtcSctp { state: RtcSctpState, endpoint: Endpoint, @@ -46,6 +49,7 @@ pub(crate) struct RtcSctp { last_now: Instant, client: bool, remote_max_message_size: u32, + max_buffered_across_streams: usize, snap_enabled: bool, snap_init: Option, #[cfg(test)] @@ -279,6 +283,7 @@ impl RtcSctp { last_now: Instant::now(), // placeholder until init() client: false, remote_max_message_size: DEFAULT_REMOTE_MAX_MESSAGE_SIZE, + max_buffered_across_streams: DEFAULT_MAX_BUFFERED_ACROSS_STREAMS, snap_enabled: false, snap_init: None, #[cfg(test)] @@ -323,6 +328,12 @@ impl RtcSctp { self.remote_max_message_size = max_msg_size; } + let bufsize: usize = + BUFFER_MULTIPLIER.saturating_mul(self.remote_max_message_size) as usize; + if self.max_buffered_across_streams < bufsize { + self.max_buffered_across_streams = bufsize; + } + if let Some(snap_data) = sctp_init_data { // SNAP path: both local and remote INIT chunks must be present. if snap_data.local_init.is_none() || snap_data.remote_init.is_none() { @@ -341,6 +352,7 @@ impl RtcSctp { .connect(config, self.fake_addr) .map_err(|e| SctpError::Proto(ProtoError::Other(e.to_string())))?; assoc.set_max_send_message_size(self.remote_max_message_size); + self.handle = handle; self.assoc = Some(assoc); @@ -569,7 +581,7 @@ impl RtcSctp { }) .sum(); - MAX_BUFFERED_ACROSS_STREAMS - total + self.max_buffered_across_streams - total } pub fn write(&mut self, id: u16, binary: bool, buf: &[u8]) -> Result { @@ -1008,7 +1020,6 @@ impl RtcSctp { .and_then(|s| s.config.as_ref()) } - #[cfg(test)] pub(crate) fn remote_max_message_size(&self) -> u32 { self.remote_max_message_size }