From 39171ee196c9f579f55ae5310fd5bb64e9149eca Mon Sep 17 00:00:00 2001 From: gyanano <1624055384@qq.com> Date: Wed, 2 Sep 2026 01:43:16 -0700 Subject: [PATCH] refactor(serial): slim read loop onto FrameSegmenter with hard frame cap RFC #3 Step 2: - Replace the four duplicated frame-emission blocks in connect() with a single emit_frame closure driven by FrameSegmenter (read loop ~330 -> ~120 lines). - Add max_frame_bytes hard cap (default 64 KiB): continuous streams with no delimiter are cut into cap-sized chunks during feed instead of accumulating unboundedly until an idle gap. Delimiters within the cap window still win. This is the only intentional behavior change. - Raw file tap stays before segmentation; read-error silent death pinned as Step 3 scope. Tests: 32/32 green (3 new framing unit tests + 1 scripted-port harness test for hard-cap cutting). cargo build warning-free. --- src-tauri/src/framing.rs | 111 ++++++++++-- src-tauri/src/serial_manager.rs | 296 ++++++++------------------------ 2 files changed, 170 insertions(+), 237 deletions(-) diff --git a/src-tauri/src/framing.rs b/src-tauri/src/framing.rs index 6546781..e0da76d 100644 --- a/src-tauri/src/framing.rs +++ b/src-tauri/src/framing.rs @@ -8,21 +8,37 @@ //! duplicated emission blocks over to this component; until then the //! segmenter is exercised only by tests. -use crate::types::{FrameDelimiter, FrameSegmentationConfig, FrameSegmentationMode}; +use crate::types::{FrameSegmentationConfig, FrameSegmentationMode}; use std::time::{Duration, Instant}; +/// Hard cap on frame size (RFC #3): a frame may never exceed this many +/// bytes. Continuous streams with no delimiter are cut into cap-sized +/// chunks during `feed`, so memory stays bounded and every frame carries a +/// size guarantee. 64 KiB matches the planned IPC batch budget. +pub const DEFAULT_MAX_FRAME_BYTES: usize = 64 * 1024; + /// Bytes in, frames out. The caller drives it with explicit timestamps so /// tests never need to sleep. pub struct FrameSegmenter { config: FrameSegmentationConfig, + max_frame_bytes: usize, buffer: Vec, last_data_time: Instant, } impl FrameSegmenter { pub fn new(config: FrameSegmentationConfig, now: Instant) -> Self { + Self::with_max_frame_bytes(config, now, DEFAULT_MAX_FRAME_BYTES) + } + + pub fn with_max_frame_bytes( + config: FrameSegmentationConfig, + now: Instant, + max_frame_bytes: usize, + ) -> Self { Self { config, + max_frame_bytes, buffer: Vec::new(), last_data_time: now, } @@ -35,32 +51,46 @@ impl FrameSegmenter { self.config = config; } - pub fn config(&self) -> &FrameSegmentationConfig { - &self.config - } - /// Feed bytes just read from the port. Returns frames closed by /// delimiter processing — delimiter bytes are included in the frame, /// matching the legacy behavior. Delimiter processing only happens in - /// Combined mode; in Timeout mode everything waits for `flush_if_idle`. + /// Combined mode; in Timeout mode everything waits for `flush_if_idle` + /// or the hard cap. + /// + /// Hard cap (new in Step 2, replaces legacy unbounded growth): a + /// delimiter only closes a frame if the match lies fully inside the + /// first `max_frame_bytes` of the buffer; beyond that the buffer is cut + /// into cap-sized chunks. A delimiter straddling the cap boundary can + /// therefore be split — same family as the pinned CRLF-across-reads + /// quirk, deterministic and bounded. Invariant on return: + /// `buffer.len() < max_frame_bytes`. pub fn feed(&mut self, bytes: &[u8], now: Instant) -> Vec> { self.buffer.extend_from_slice(bytes); self.last_data_time = now; + let delimiter = self.config.delimiter.to_bytes(); + let combined = self.config.mode == FrameSegmentationMode::Combined; let mut frames = Vec::new(); - if self.config.mode == FrameSegmentationMode::Combined { - if self.config.delimiter.is_any_newline() { - while let Some((pos, len)) = find_any_newline(&self.buffer) { - let frame_end = pos + len; - frames.push(self.buffer.drain(..frame_end).collect()); - } - } else { - let delimiter = self.config.delimiter.to_bytes(); - while let Some(pos) = find_delimiter(&self.buffer, &delimiter) { - let frame_end = pos + delimiter.len(); - frames.push(self.buffer.drain(..frame_end).collect()); + loop { + if combined { + let hit = { + let window = &self.buffer[..self.buffer.len().min(self.max_frame_bytes)]; + if self.config.delimiter.is_any_newline() { + find_any_newline(window) + } else { + find_delimiter(window, &delimiter).map(|pos| (pos, delimiter.len())) + } + }; + if let Some((pos, len)) = hit { + frames.push(self.buffer.drain(..pos + len).collect()); + continue; } } + if self.buffer.len() >= self.max_frame_bytes { + frames.push(self.buffer.drain(..self.max_frame_bytes).collect()); + continue; + } + break; } frames } @@ -131,6 +161,7 @@ pub(crate) fn find_any_newline(data: &[u8]) -> Option<(usize, usize)> { #[cfg(test)] mod tests { use super::*; + use crate::types::FrameDelimiter; fn combined(delimiter: FrameDelimiter) -> FrameSegmentationConfig { FrameSegmentationConfig { @@ -262,4 +293,50 @@ mod tests { Some(b"ab".to_vec()) ); } + + #[test] + fn hard_cap_cuts_continuous_stream_without_waiting_for_idle() { + let t0 = Instant::now(); + let mut seg = + FrameSegmenter::with_max_frame_bytes(timeout_mode(), t0, 1024); + // 2500 bytes with no idle gap: two full chunks cut immediately, + // residue waits for the timeout flush. + let frames = seg.feed(&vec![b'x'; 2500], t0); + assert_eq!(frames.len(), 2); + assert!(frames.iter().all(|f| f.len() == 1024)); + assert_eq!(seg.pending().len(), 452); + assert_eq!( + seg.flush_if_idle(t0 + Duration::from_millis(11)), + Some(vec![b'x'; 452]) + ); + } + + #[test] + fn delimiter_within_cap_window_wins_over_hard_cut() { + let t0 = Instant::now(); + let mut seg = FrameSegmenter::with_max_frame_bytes( + combined(FrameDelimiter::LF), + t0, + 8, + ); + // LF inside the first 8 bytes closes a short frame; rest pends. + let frames = seg.feed(b"ab\ncdefgh", t0); + assert_eq!(frames, vec![b"ab\n".to_vec()]); + assert_eq!(seg.pending(), b"cdefgh"); + } + + #[test] + fn delimiter_beyond_cap_window_gets_hard_cut() { + let t0 = Instant::now(); + let mut seg = FrameSegmenter::with_max_frame_bytes( + combined(FrameDelimiter::LF), + t0, + 8, + ); + // LF at position 8 is outside the 8-byte window: hard cut first, + // then the lone LF frames on its own (cap-boundary split, pinned). + let frames = seg.feed(b"abcdefgh\n", t0); + assert_eq!(frames, vec![b"abcdefgh".to_vec(), b"\n".to_vec()]); + assert!(seg.pending().is_empty()); + } } diff --git a/src-tauri/src/serial_manager.rs b/src-tauri/src/serial_manager.rs index 35b4bbf..8d9f97e 100644 --- a/src-tauri/src/serial_manager.rs +++ b/src-tauri/src/serial_manager.rs @@ -1,4 +1,4 @@ -use crate::framing::{find_any_newline, find_delimiter}; +use crate::framing::FrameSegmenter; use crate::types::*; use anyhow::{anyhow, Result}; use chrono::Utc; @@ -188,9 +188,57 @@ impl SerialManager { let mut read_port = port.try_clone()?; thread::spawn(move || { - let mut buffer = [0; 1024]; - let mut accumulated_data = Vec::new(); - let mut last_data_time = Instant::now(); + let mut read_buffer = [0u8; 1024]; + let initial_config = frame_segmentation_config.lock() + .map(|guard| guard.clone()) + .unwrap_or_default(); + let mut segmenter = FrameSegmenter::new(initial_config, Instant::now()); + + // Single frame-emission path (replaces the four duplicated + // blocks): text recording -> display formatting -> log buffer -> stats. + let emit_frame = |frame_data: Vec, disp_settings: &DisplaySettings| { + // Write to text recording file with timestamp and RX label + if let Ok(mut guard) = text_file.lock() { + if let Some(ref mut file) = *guard { + let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); + let timestamp = format_timestamp_with_offset(tz_offset); + let text = String::from_utf8_lossy(&frame_data); + let _ = writeln!(file, "[{}] RX: {}", timestamp, text); + } + } + + // Format display text and timestamp based on current settings + let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); + let display_text = format_data_for_display(&frame_data, disp_settings); + let timestamp_formatted = if disp_settings.show_timestamps { + Some(format_timestamp_with_offset(tz_offset)) + } else { + None + }; + + let data_len = frame_data.len() as u64; + let log_entry = LogEntry { + timestamp: Utc::now(), + direction: Direction::Received, + data: frame_data, + format: DataFormat::Text, + port_name: port_name_clone.clone(), + display_text, + timestamp_formatted, + }; + + if let Ok(mut logs_guard) = logs.lock() { + logs_guard.push_back(log_entry); + let max_entries = *max_log_entries.lock().unwrap_or_else(|e| e.into_inner()); + while logs_guard.len() > max_entries { + logs_guard.pop_front(); + } + } + + if let Ok(mut stats_guard) = stats.lock() { + stats_guard.bytes_received += data_len; + } + }; loop { // Check shutdown flag @@ -199,246 +247,41 @@ impl SerialManager { break; } - // Get current segmentation config + // Get current segmentation config (legacy re-read each iteration) let seg_config = frame_segmentation_config.lock() .map(|guard| guard.clone()) .unwrap_or_default(); - let timeout_duration = Duration::from_millis(seg_config.timeout_ms); + segmenter.set_config(seg_config); // Get current display settings for formatting let disp_settings = display_settings.lock() .map(|guard| guard.clone()) .unwrap_or_default(); - match read_port.read(&mut buffer) { + match read_port.read(&mut read_buffer) { Ok(bytes_read) if bytes_read > 0 => { - let received_bytes = &buffer[..bytes_read]; - accumulated_data.extend_from_slice(received_bytes); - last_data_time = Instant::now(); + let received_bytes = &read_buffer[..bytes_read]; - // Write to raw recording file (raw bytes, no framing) + // Write to raw recording file (raw bytes, pre-framing tap) if let Ok(mut guard) = raw_file.lock() { if let Some(ref mut file) = *guard { let _ = file.write_all(received_bytes); } } - // Check for delimiter-based segmentation (only in Combined mode) - if seg_config.mode == FrameSegmentationMode::Combined { - - // Handle AnyNewline specially - it matches \r, \n, or \r\n as single delimiter - if seg_config.delimiter.is_any_newline() { - while let Some((pos, len)) = find_any_newline(&accumulated_data) { - let frame_end = pos + len; - let frame_data: Vec = accumulated_data.drain(..frame_end).collect(); - let data_len = frame_data.len(); - - // Write to text recording file with timestamp and RX label - if let Ok(mut guard) = text_file.lock() { - if let Some(ref mut file) = *guard { - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let timestamp = format_timestamp_with_offset(tz_offset); - let text = String::from_utf8_lossy(&frame_data); - let _ = writeln!(file, "[{}] RX: {}", timestamp, text); - } - } - - // Format display text and timestamp based on current settings - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let display_text = format_data_for_display(&frame_data, &disp_settings); - let timestamp_formatted = if disp_settings.show_timestamps { - Some(format_timestamp_with_offset(tz_offset)) - } else { - None - }; - - let log_entry = LogEntry { - timestamp: Utc::now(), - direction: Direction::Received, - data: frame_data, - format: DataFormat::Text, - port_name: port_name_clone.clone(), - display_text, - timestamp_formatted, - }; - - if let Ok(mut logs_guard) = logs.lock() { - logs_guard.push_back(log_entry); - let max_entries = *max_log_entries.lock().unwrap_or_else(|e| e.into_inner()); - while logs_guard.len() > max_entries { - logs_guard.pop_front(); - } - } - - if let Ok(mut stats_guard) = stats.lock() { - stats_guard.bytes_received += data_len as u64; - } - } - } else { - // Standard delimiter matching - let delimiter_bytes = seg_config.delimiter.to_bytes(); - - // Process all complete frames in accumulated data - while let Some(pos) = find_delimiter(&accumulated_data, &delimiter_bytes) { - let frame_end = pos + delimiter_bytes.len(); - let frame_data: Vec = accumulated_data.drain(..frame_end).collect(); - let data_len = frame_data.len(); - - // Write to text recording file with timestamp and RX label - if let Ok(mut guard) = text_file.lock() { - if let Some(ref mut file) = *guard { - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let timestamp = format_timestamp_with_offset(tz_offset); - let text = String::from_utf8_lossy(&frame_data); - let _ = writeln!(file, "[{}] RX: {}", timestamp, text); - } - } - - // Format display text and timestamp based on current settings - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let display_text = format_data_for_display(&frame_data, &disp_settings); - let timestamp_formatted = if disp_settings.show_timestamps { - Some(format_timestamp_with_offset(tz_offset)) - } else { - None - }; - - let log_entry = LogEntry { - timestamp: Utc::now(), - direction: Direction::Received, - data: frame_data, - format: DataFormat::Text, - port_name: port_name_clone.clone(), - display_text, - timestamp_formatted, - }; - - if let Ok(mut logs_guard) = logs.lock() { - logs_guard.push_back(log_entry); - let max_entries = *max_log_entries.lock().unwrap_or_else(|e| e.into_inner()); - while logs_guard.len() > max_entries { - logs_guard.pop_front(); - } - } - - if let Ok(mut stats_guard) = stats.lock() { - stats_guard.bytes_received += data_len as u64; - } - } - } + for frame in segmenter.feed(received_bytes, Instant::now()) { + emit_frame(frame, &disp_settings); } } Ok(_) => { - // Check if we should flush accumulated data based on timeout - let should_flush_timeout = - (seg_config.mode == FrameSegmentationMode::Timeout || - seg_config.mode == FrameSegmentationMode::Combined) && - !accumulated_data.is_empty() && - last_data_time.elapsed() > timeout_duration; - - if should_flush_timeout { - let data_len = accumulated_data.len(); - - // Write to text recording file with timestamp and RX label - if let Ok(mut guard) = text_file.lock() { - if let Some(ref mut file) = *guard { - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let timestamp = format_timestamp_with_offset(tz_offset); - let text = String::from_utf8_lossy(&accumulated_data); - let _ = writeln!(file, "[{}] RX: {}", timestamp, text); - } - } - - // Format display text and timestamp based on current settings - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let display_text = format_data_for_display(&accumulated_data, &disp_settings); - let timestamp_formatted = if disp_settings.show_timestamps { - Some(format_timestamp_with_offset(tz_offset)) - } else { - None - }; - - let log_entry = LogEntry { - timestamp: Utc::now(), - direction: Direction::Received, - data: accumulated_data.clone(), - format: DataFormat::Text, - port_name: port_name_clone.clone(), - display_text, - timestamp_formatted, - }; - - if let Ok(mut logs_guard) = logs.lock() { - logs_guard.push_back(log_entry); - let max_entries = *max_log_entries.lock().unwrap_or_else(|e| e.into_inner()); - while logs_guard.len() > max_entries { - logs_guard.pop_front(); - } - } - - // Update received bytes statistics - if let Ok(mut stats_guard) = stats.lock() { - stats_guard.bytes_received += data_len as u64; - } - - accumulated_data.clear(); + if let Some(frame) = segmenter.flush_if_idle(Instant::now()) { + emit_frame(frame, &disp_settings); } thread::sleep(Duration::from_millis(1)); } Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut => { - // Check if we should flush accumulated data on timeout - let should_flush_timeout = - (seg_config.mode == FrameSegmentationMode::Timeout || - seg_config.mode == FrameSegmentationMode::Combined) && - !accumulated_data.is_empty() && - last_data_time.elapsed() > timeout_duration; - - if should_flush_timeout { - let data_len = accumulated_data.len(); - - // Write to text recording file with timestamp and RX label - if let Ok(mut guard) = text_file.lock() { - if let Some(ref mut file) = *guard { - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let timestamp = format_timestamp_with_offset(tz_offset); - let text = String::from_utf8_lossy(&accumulated_data); - let _ = writeln!(file, "[{}] RX: {}", timestamp, text); - } - } - - // Format display text and timestamp based on current settings - let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner()); - let display_text = format_data_for_display(&accumulated_data, &disp_settings); - let timestamp_formatted = if disp_settings.show_timestamps { - Some(format_timestamp_with_offset(tz_offset)) - } else { - None - }; - - let log_entry = LogEntry { - timestamp: Utc::now(), - direction: Direction::Received, - data: accumulated_data.clone(), - format: DataFormat::Text, - port_name: port_name_clone.clone(), - display_text, - timestamp_formatted, - }; - - if let Ok(mut logs_guard) = logs.lock() { - logs_guard.push_back(log_entry); - let max_entries = *max_log_entries.lock().unwrap_or_else(|e| e.into_inner()); - while logs_guard.len() > max_entries { - logs_guard.pop_front(); - } - } - - // Update received bytes statistics - if let Ok(mut stats_guard) = stats.lock() { - stats_guard.bytes_received += data_len as u64; - } - - accumulated_data.clear(); + if let Some(frame) = segmenter.flush_if_idle(Instant::now()) { + emit_frame(frame, &disp_settings); } thread::sleep(Duration::from_millis(1)); } @@ -1480,4 +1323,17 @@ mod tests { assert!(manager.get_logs().is_empty()); // pending "abc" lost, pinned manager.disconnect().unwrap(); } + + #[test] + fn golden_continuous_stream_hard_capped_frames() { + // 150 KiB of back-to-back data with no idle gap: the hard cap must + // cut 64 KiB frames without waiting for a timeout, and the residue + // flushes once the stream goes quiet. + let script: Vec = (0..150) + .map(|_| ScriptEvent::Bytes(vec![b'x'; 1024])) + .collect(); + let logs = run_script(script, seg_timeout(), 3); + let sizes: Vec = logs.iter().map(|l| l.data.len()).collect(); + assert_eq!(sizes, vec![65536, 65536, 22528]); + } } \ No newline at end of file