@@ -60,10 +60,14 @@ fn opus_head_codec_private(sample_rate: u32, channels: u32) -> Result<[u8; 19],
6060/// A shared, thread-safe buffer that wraps a Cursor for WebM writing.
6161/// This allows us to stream out data as it's written while still supporting Seek.
6262///
63- /// Uses a sliding window approach to prevent unbounded memory growth:
64- /// - Keeps a configurable window of recent data for WebM library seeks
65- /// - Discards old data beyond the window that's been sent
66- /// - Tracks a base_offset for proper position calculations after discarding
63+ /// Supports two buffering modes:
64+ ///
65+ /// - **Streaming (non-seek)**: Bytes are drained on every `take_data()` call.
66+ /// This mode is intended for `Writer::new_non_seek` and avoids copying.
67+ /// - **Seek window**: Keeps a configurable window of recent data for WebM library seeks
68+ /// and trims old data that has already been sent.
69+ ///
70+ /// The node selects the appropriate mode based on `WebMStreamingMode`.
6771#[ derive( Clone ) ]
6872struct SharedPacketBuffer {
6973 cursor : Arc < Mutex < Cursor < Vec < u8 > > > > ,
@@ -84,9 +88,14 @@ impl SharedPacketBuffer {
8488 }
8589 }
8690
87- fn new ( ) -> Self {
88- // Default 1MB window (enough for ~6 seconds of 128kbps audio)
89- Self :: new_with_window ( 1024 * 1024 )
91+ /// Create a non-seek streaming buffer.
92+ ///
93+ /// This is designed for `Writer::new_non_seek` in live streaming mode. Since the writer
94+ /// does not seek/backpatch, we can drain bytes out by moving the underlying `Vec<u8>`
95+ /// (no copy) and reset the cursor to keep memory bounded.
96+ fn new_streaming ( ) -> Self {
97+ // window_size=0 is treated as "drain everything on take_data"
98+ Self :: new_with_window ( 0 )
9099 }
91100
92101 /// Takes any new data written since the last call, and trims old data beyond the window.
@@ -108,30 +117,52 @@ impl SharedPacketBuffer {
108117 let base = * base_offset_guard;
109118
110119 let result = if current_len > last_sent {
111- // Copy only the new data since last send
112- let new_data = Bytes :: copy_from_slice ( & vec[ last_sent..current_len] ) ;
113- * last_sent_guard = current_len;
114-
115- // Trim old data if buffer exceeds window size
116- if current_len > self . window_size {
117- let trim_amount = current_len - self . window_size ;
118- // Keep the last window_size bytes
119- let remaining = vec. split_off ( trim_amount) ;
120- * vec = remaining;
121- // Update base offset to reflect discarded data
122- * base_offset_guard = base + trim_amount;
123- // Adjust last_sent and cursor position
124- * last_sent_guard = self . window_size ;
125- buffer_guard. set_position ( self . window_size as u64 ) ;
126-
127- tracing:: debug!(
128- "Trimmed {} bytes from WebM buffer, new base_offset: {}" ,
129- trim_amount,
130- * base_offset_guard
131- ) ;
132- }
120+ if self . window_size == 0 {
121+ // Streaming mode (non-seek): drain everything written so far without copying.
122+ //
123+ // This avoids two major sources of allocation churn in DHAT profiles:
124+ // - copying out incremental slices on every flush
125+ // - repeatedly trimming a sliding window with `split_off` (copies the window)
126+ let data_vec = std:: mem:: take ( vec) ;
127+ // Advance base_offset so Seek::Start can clamp consistently if it ever happens.
128+ * base_offset_guard = base + current_len;
129+ * last_sent_guard = 0 ;
130+ buffer_guard. set_position ( 0 ) ;
131+ Some ( Bytes :: from ( data_vec) )
132+ } else if self . window_size == usize:: MAX && last_sent == 0 {
133+ // File mode: nothing has been sent yet, so move the entire buffer out.
134+ // The segment is finalized before this is called, so no more writes/seeks occur.
135+ let data_vec = std:: mem:: take ( vec) ;
136+ * base_offset_guard = base + current_len;
137+ * last_sent_guard = 0 ;
138+ buffer_guard. set_position ( 0 ) ;
139+ Some ( Bytes :: from ( data_vec) )
140+ } else {
141+ // Seek-window mode: copy incremental bytes while retaining a backwards-seek window.
142+ let new_data = Bytes :: copy_from_slice ( & vec[ last_sent..current_len] ) ;
143+ * last_sent_guard = current_len;
144+
145+ // Trim old data if buffer exceeds window size.
146+ if current_len > self . window_size {
147+ let trim_amount = current_len - self . window_size ;
148+ // Keep the last window_size bytes.
149+ let remaining = vec. split_off ( trim_amount) ;
150+ * vec = remaining;
151+ // Update base offset to reflect discarded data.
152+ * base_offset_guard = base + trim_amount;
153+ // Adjust last_sent and cursor position.
154+ * last_sent_guard = self . window_size ;
155+ buffer_guard. set_position ( self . window_size as u64 ) ;
156+
157+ tracing:: debug!(
158+ "Trimmed {} bytes from WebM buffer, new base_offset: {}" ,
159+ trim_amount,
160+ * base_offset_guard
161+ ) ;
162+ }
133163
134- Some ( new_data)
164+ Some ( new_data)
165+ }
135166 } else {
136167 None
137168 } ;
@@ -286,11 +317,11 @@ impl ProcessorNode for WebMMuxerNode {
286317 // Stats tracking
287318 let mut stats_tracker = NodeStatsTracker :: new ( node_name. clone ( ) , context. stats_tx . clone ( ) ) ;
288319
289- // In Live mode we only need a small sliding window to support any internal backtracking
290- // while continuously streaming bytes out; in File mode we must keep the whole buffer
320+ // In Live mode we use a non-seek writer, so we can drain bytes out without keeping
321+ // any history (zero-copy streaming). In File mode we must keep the whole buffer
291322 // because we only emit bytes once the segment is finalized.
292323 let shared_buffer = match self . config . streaming_mode {
293- WebMStreamingMode :: Live => SharedPacketBuffer :: new ( ) ,
324+ WebMStreamingMode :: Live => SharedPacketBuffer :: new_streaming ( ) ,
294325 WebMStreamingMode :: File => SharedPacketBuffer :: new_with_window ( usize:: MAX ) ,
295326 } ;
296327
0 commit comments