From 85a3bfcb1bcba0ca6b3a1c0aaf98285eba6fef6c Mon Sep 17 00:00:00 2001 From: JoeAllen <2494136784@qq.com> Date: Tue, 29 Sep 2026 17:00:34 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E6=B6=88=E6=81=AF=E5=86=85=E5=AD=98?= =?UTF-8?q?=E6=8C=89=E5=AD=97=E8=8A=82=E9=A2=84=E7=AE=97=EF=BC=8C=E6=9D=9C?= =?UTF-8?q?=E7=BB=9D=E5=A4=A7=E6=8A=A5=E6=96=87=E6=B5=81=E6=92=91=E7=88=86?= =?UTF-8?q?=E5=86=85=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 MessageRing：条数上限 + 每连接 32MB 字节预算双重驱逐 （从最旧丢弃，均摊 O(1)）；预分配容量按设置钳制在 512 - 单条 payload 保留上限 256KB → 128KB；MqttRecord::retained_bytes 计入文本与原始字节，供预算驱逐与资源监控展示 - 清空消息统一走 clear_messages；设置页标注字节预算兜底说明 - 新增单元测试：条数驱逐、字节预算、超预算单条保留、驱逐顺序 --- gpui-app/src/model.rs | 140 ++++++++++++++++++++++++++++- gpui-app/src/ui/app.rs | 28 +++--- gpui-app/src/ui/connection_view.rs | 2 +- gpui-app/src/ui/settings_dialog.rs | 11 ++- 4 files changed, 164 insertions(+), 17 deletions(-) diff --git a/gpui-app/src/model.rs b/gpui-app/src/model.rs index 36a1d73..0f63248 100644 --- a/gpui-app/src/model.rs +++ b/gpui-app/src/model.rs @@ -6,6 +6,8 @@ use std::sync::Arc; use serde::{Deserialize, Serialize}; +use std::collections::VecDeque; + // ─── 连接配置 ──────────────────────────────────────────────────────────────── /// MQTT 协议版本。 @@ -451,9 +453,70 @@ pub struct MqttRecord { pub raw_bytes: Option>, } +impl MqttRecord { + /// 该记录保留的主要字节数（文本 + 原始字节），用于内存预算驱逐。 + /// 其余字段（主题/属性等）相对小一个数量级，不计入。 + pub fn retained_bytes(&self) -> usize { + self.payload.len() + self.raw_bytes.as_ref().map_or(0, |b| b.len()) + } +} + +/// 单连接的消息环形缓冲：条数上限 + [`MAX_CONNECTION_MESSAGE_BYTES`] +/// 字节上限双重驱逐（从最旧开始，均摊 O(1)）。 +pub struct MessageRing { + queue: VecDeque, + bytes: usize, +} + +impl MessageRing { + pub fn new(max_count: usize) -> Self { + Self { + // 预分配按较小值钳制：设置上限十万条时不一次性分配过大 + queue: VecDeque::with_capacity(max_count.clamp(16, 512)), + bytes: 0, + } + } + + /// `max_count` 每次传入以便运行时调整设置后即时生效。 + /// 设置层已校验 ≥100，这里只防 0 值导致缓冲被清空。 + pub fn push(&mut self, record: MqttRecord, max_count: usize) { + let max_count = max_count.max(1); + self.bytes += record.retained_bytes(); + while self.queue.len() >= max_count + || (self.bytes > MAX_CONNECTION_MESSAGE_BYTES && self.queue.len() > 1) + { + let Some(old) = self.queue.pop_front() else { break }; + self.bytes -= old.retained_bytes(); + } + self.queue.push_back(record); + } + + pub fn len(&self) -> usize { + self.queue.len() + } + + pub fn is_empty(&self) -> bool { + self.queue.is_empty() + } + + /// 当前保留的字节数（payload + 原始字节）。 + pub fn retained_bytes(&self) -> usize { + self.bytes + } + + pub fn iter(&self) -> std::collections::vec_deque::Iter<'_, MqttRecord> { + self.queue.iter() + } +} + /// 消息历史中单条 payload 的保留上限。超限截断，避免个别大报文 /// 长期占据历史缓冲（MQTTX 亦有类似限制）。 -pub const MAX_PAYLOAD_RETAIN: usize = 256 * 1024; +pub const MAX_PAYLOAD_RETAIN: usize = 128 * 1024; + +/// 每连接消息历史的字节预算（payload + 原始字节）。条数上限 +/// （max_messages）在大报文流下形同虚设（2000 条 × 128KB ≈ 256MB）， +/// 超出预算后从最旧开始驱逐，保证单连接消息内存有上界。 +pub const MAX_CONNECTION_MESSAGE_BYTES: usize = 32 * 1024 * 1024; /// 把原始 payload 字节转成用于历史记录的共享文本，超限截断。 /// 返回 (文本, 非文本时的原始字节, 是否截断)。截断时回退到 UTF-8 字符边界。 @@ -985,4 +1048,79 @@ mod tests { b.ssl.ca_file = "/tmp/ca.pem".into(); assert!(a.session_params_changed(&b), "TLS 配置改动应触发重连提示"); } + + fn ring_record(seq: u64, payload_len: usize, raw_len: usize) -> MqttRecord { + MqttRecord { + seq, + connection_id: "c".into(), + direction: Direction::Received, + topic: "t".into(), + payload: Arc::from("a".repeat(payload_len)), + qos: 0, + retain: false, + timestamp: 0, + user_properties: Vec::new(), + content_type: None, + response_topic: None, + correlation_data: None, + message_expiry_interval: None, + subscription_identifier: None, + payload_truncated: false, + raw_bytes: if raw_len > 0 { + Some(Arc::from(vec![0u8; raw_len])) + } else { + None + }, + } + } + + // 消息环形缓冲：条数上限正常驱逐 + #[test] + fn message_ring_enforces_count_limit() { + let mut ring = MessageRing::new(4); + for i in 0..10 { + ring.push(ring_record(i, 64, 0), 4); + } + assert_eq!(ring.len(), 4); + assert_eq!( + ring.iter().next().map(|r| r.seq), + Some(6), + "应从最旧开始驱逐" + ); + assert_eq!(ring.retained_bytes(), 4 * 64); + } + + // 消息环形缓冲：大报文流下字节预算优先于条数上限，内存有上界 + #[test] + fn message_ring_enforces_byte_budget() { + let mut ring = MessageRing::new(2000); + // 每条 256KB（payload 128KB + raw 128KB）：2000 条本会到 512MB + let per = MAX_PAYLOAD_RETAIN; + for i in 0..2000 { + ring.push(ring_record(i, per, per), 2000); + } + assert!( + ring.retained_bytes() <= MAX_CONNECTION_MESSAGE_BYTES + 2 * per, + "字节数 {} 超出预算 {}", + ring.retained_bytes(), + MAX_CONNECTION_MESSAGE_BYTES + ); + assert!( + ring.len() < 2000, + "大报文流不应保留满额条数（实际 {} 条）", + ring.len() + ); + assert_eq!(ring.iter().next().map(|r| r.seq), Some((2000 - ring.len()) as u64)); + } + + // 单条超预算的记录也至少保留最新一条（列表不能被清空） + #[test] + fn message_ring_never_empties_on_oversized_record() { + let mut ring = MessageRing::new(10); + ring.push(ring_record(0, MAX_PAYLOAD_RETAIN, MAX_PAYLOAD_RETAIN), 10); + assert_eq!(ring.len(), 1); + ring.push(ring_record(1, 64, 0), 10); + assert_eq!(ring.len(), 2); + assert_eq!(ring.iter().next().map(|r| r.seq), Some(0)); + } } diff --git a/gpui-app/src/ui/app.rs b/gpui-app/src/ui/app.rs index 1f1c4ae..174e484 100644 --- a/gpui-app/src/ui/app.rs +++ b/gpui-app/src/ui/app.rs @@ -24,8 +24,8 @@ use gpui_kit::{ use crate::aliyun::AliyunPreset; use crate::model::{ AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry, - LogLevel, MqttRecord, PublishParams, PublishPreset, SubscribeOptions, Subscription, - ThemeModePref, retained_payload, + LogLevel, MqttRecord, MessageRing, PublishParams, PublishPreset, SubscribeOptions, + Subscription, ThemeModePref, retained_payload, }; use crate::mqtt::{EngineEvent, MqttEngine}; use crate::store::Storage; @@ -84,8 +84,8 @@ pub struct MqttXApp { // ── 运行时状态 ── pub statuses: HashMap, pub errors: HashMap, - /// 每连接消息环形缓冲：键为驻留后的连接 ID，容量受 max_messages 约束 - pub messages: HashMap, VecDeque>, + /// 每连接消息环形缓冲：键为驻留后的连接 ID，条数 + 字节预算双重约束 + pub messages: HashMap, MessageRing>, pub logs: VecDeque, pub seq: u64, @@ -473,16 +473,16 @@ impl MqttXApp { } fn push_message(&mut self, connection_id: Arc, record: MqttRecord) { - let cap = self.settings.max_messages.max(100); - let list = self - .messages + let cap = self.settings.max_messages; + self.messages .entry(connection_id) - .or_insert_with(|| VecDeque::with_capacity(cap)); - // 环形缓冲：满则丢最旧，均摊 O(1)（原来的 Vec::drain 是 O(n) 搬移） - while list.len() >= cap { - list.pop_front(); - } - list.push_back(record); + .or_insert_with(|| MessageRing::new(cap)) + .push(record, cap); + } + + /// 清空指定连接的消息（内存缓冲随 drop 释放）。 + pub fn clear_messages(&mut self, id: &str) { + self.messages.remove(id); } fn push_log(&mut self, entry: LogEntry) { @@ -603,7 +603,7 @@ impl MqttXApp { self.engine.close(id, false); self.connections.retain(|c| c.id != id); self.subscriptions.retain(|s| s.connection_id != id); - self.messages.remove(id); + self.clear_messages(id); self.statuses.remove(id); self.errors.remove(id); self.views.remove(id); diff --git a/gpui-app/src/ui/connection_view.rs b/gpui-app/src/ui/connection_view.rs index bb92f19..84c4597 100644 --- a/gpui-app/src/ui/connection_view.rs +++ b/gpui-app/src/ui/connection_view.rs @@ -1765,7 +1765,7 @@ impl ConnectionView { view.detail_format = DetailFormat::Auto; if let Some(app) = view.app.upgrade() { app.update(cx, |app, cx| { - app.messages.remove(clear_id.as_str()); + app.clear_messages(&clear_id); cx.notify(); }); } diff --git a/gpui-app/src/ui/settings_dialog.rs b/gpui-app/src/ui/settings_dialog.rs index f3b46bd..a8cd629 100644 --- a/gpui-app/src/ui/settings_dialog.rs +++ b/gpui-app/src/ui/settings_dialog.rs @@ -17,7 +17,7 @@ use gpui_kit::{ Styled as _, Window, }; -use crate::model::{AppSettings, ThemeModePref}; +use crate::model::{AppSettings, ThemeModePref, MAX_CONNECTION_MESSAGE_BYTES}; use crate::mqtt::MqttEngine; use crate::ui::app::MqttXApp; use crate::ui::widgets::{field, make_select, OptionDelegate}; @@ -388,6 +388,15 @@ impl Render for SettingsDialog { "每条连接内存中保留的消息条数（100~100000）", Input::new(&self.max_messages), )) + .child( + div() + .text_xs() + .text_color(cx.theme().muted_foreground) + .child(format!( + "另有每连接 {}MB 的字节预算兜底：大报文流会优先按字节从最旧开始驱逐", + MAX_CONNECTION_MESSAGE_BYTES / (1024 * 1024) + )), + ) .child( h_flex() .justify_between()