From 96865c0588df77eebc4035500f3a9461bbbf2a45 Mon Sep 17 00:00:00 2001 From: JoeAllen <2494136784@qq.com> Date: Tue, 29 Sep 2026 01:07:32 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E6=B7=B1=E5=BA=A6=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=86=85=E5=AD=98=E5=8D=A0=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 消息缓冲 Vec → VecDeque：原来溢出时 drain(0..n) 每条消息 O(n) 搬移，高频消息下产生大量拷贝；环形缓冲均摊 O(1)，并按配置上限 预分配避免反复扩容 - connection_id/topic/payload 改 Arc：引擎侧主题驻留 （10k 上限，超出整体重建），同主题数千条记录只存一份； 消息列表每帧克隆最多 300 条用于渲染，payload 共享后零拷贝 - 单条 payload 历史保留上限 256KB（retained_payload，UTF-8 边界安全截断），超大报文不再长期占据环形缓冲；消息行显示 「已截断」徽章；发布记录改用实际发送字节（修 Base64/Hex 历史内容与真实报文不一致的隐患） - live 测试适配 Arc；补截断边界单元测试（22 个 lib 测试 + 2 个 live 测试通过） --- gpui-app/src/model.rs | 60 ++++++++++++++++++++++++++++-- gpui-app/src/mqtt.rs | 48 ++++++++++++++++++------ gpui-app/src/ui/app.rs | 33 +++++++++------- gpui-app/src/ui/connection_view.rs | 26 +++++++++---- gpui-app/tests/engine_live.rs | 2 +- 5 files changed, 134 insertions(+), 35 deletions(-) diff --git a/gpui-app/src/model.rs b/gpui-app/src/model.rs index 72232b0..e781a0d 100644 --- a/gpui-app/src/model.rs +++ b/gpui-app/src/model.rs @@ -2,6 +2,8 @@ //! //! 本层不依赖 GPUI，可被持久化层与 MQTT 引擎共用。 +use std::sync::Arc; + use serde::{Deserialize, Serialize}; // ─── 连接配置 ──────────────────────────────────────────────────────────────── @@ -410,13 +412,17 @@ impl PayloadFormat { /// 一条收到/发出的消息记录。payload 以 UTF-8 文本保存（非文本按 lossy 显示）。 +/// +/// 内存优化：`connection_id`/`topic` 用 `Arc`——同一连接内主题高度重复， +/// 驻留后数千条记录只保留一份字符串。 #[derive(Debug, Clone)] pub struct MqttRecord { pub seq: u64, - pub connection_id: String, + pub connection_id: Arc, pub direction: Direction, - pub topic: String, - pub payload: String, + pub topic: Arc, + /// Arc 共享：消息列表渲染每帧克隆最多数百条记录，零拷贝 + pub payload: Arc, pub qos: u8, pub retain: bool, /// unix 毫秒 @@ -427,6 +433,29 @@ pub struct MqttRecord { pub correlation_data: Option, pub message_expiry_interval: Option, pub subscription_identifier: Option, + /// payload 超过 [`MAX_PAYLOAD_RETAIN`] 被截断 + pub payload_truncated: bool, +} + +/// 消息历史中单条 payload 的保留上限。超限截断，避免个别大报文 +/// 长期占据历史缓冲（MQTTX 亦有类似限制）。 +pub const MAX_PAYLOAD_RETAIN: usize = 256 * 1024; + +/// 把原始 payload 字节转成用于历史记录的共享文本，超限截断。 +/// 返回 (Arc, 是否截断)。截断时回退到 UTF-8 字符边界。 +pub fn retained_payload(raw: &[u8]) -> (Arc, bool) { + if raw.len() <= MAX_PAYLOAD_RETAIN { + (Arc::from(String::from_utf8_lossy(raw).as_ref()), false) + } else { + let mut end = MAX_PAYLOAD_RETAIN; + while end > 0 && (raw[end] & 0xC0) == 0x80 { + end -= 1; + } + ( + Arc::from(String::from_utf8_lossy(&raw[..end]).as_ref()), + true, + ) + } } // ─── 发布参数 / 发布预设 ────────────────────────────────────────────────────── @@ -863,4 +892,29 @@ mod tests { // Subscription::new 同样产出默认启用的订阅 assert!(Subscription::new("c1", "t", 0).enabled); } + + // payload 截断：上限内不截断；超出时截断且不切坏 UTF-8 字符 + #[test] + fn retained_payload_caps_and_preserves_utf8() { + let small = b"hello"; + let (p, trunc) = retained_payload(small); + assert_eq!(&*p, "hello"); + assert!(!trunc); + + // 中文每字符 3 字节，让上限落在字符中间（256K 不是 3 的倍数） + let big: Vec = "中".repeat(MAX_PAYLOAD_RETAIN).into_bytes(); + let (p, trunc) = retained_payload(&big); + assert!(trunc, "超过上限必须截断"); + assert!(p.len() <= MAX_PAYLOAD_RETAIN); + assert_eq!(p.chars().last(), Some('中'), "不能切坏 UTF-8 字符"); + assert!( + big.starts_with(p.as_bytes()), + "截断内容必须是原数据前缀" + ); + + // 恰好上限不截断 + let exact = vec![b'a'; MAX_PAYLOAD_RETAIN]; + let (_, trunc) = retained_payload(&exact); + assert!(!trunc); + } } diff --git a/gpui-app/src/mqtt.rs b/gpui-app/src/mqtt.rs index be426d0..b37c94b 100644 --- a/gpui-app/src/mqtt.rs +++ b/gpui-app/src/mqtt.rs @@ -37,11 +37,11 @@ pub enum EngineEvent { }, /// 收到消息 Message(MqttRecord), - /// 一条本地发出的消息（已成功交给客户端） + /// 一条本地发出的消息（已成功交给客户端）。payload 为实际发送的字节。 Published { connection_id: String, topic: String, - payload: String, + payload: Vec, qos: u8, retain: bool, content_type: Option, @@ -77,6 +77,8 @@ struct ConnHandle { pub struct MqttEngine { runtime: tokio::runtime::Runtime, conns: Mutex>, + /// 主题/连接 ID 驻留缓存：同一连接内主题高度重复，驻留让消息记录共享字符串 + topics: Mutex>>, tx: EventSender, seq: Arc, } @@ -96,6 +98,7 @@ impl MqttEngine { Arc::new(Self { runtime, conns: Mutex::new(HashMap::new()), + topics: Mutex::new(HashMap::new()), tx, seq: Arc::new(AtomicU64::new(1)), }) @@ -272,6 +275,8 @@ impl MqttEngine { .raw_bytes .clone() .unwrap_or_else(|| params.payload.as_bytes().to_vec()); + // 实际发送的字节留给 Published 事件（记录需要真实内容） + let sent_payload = payload.clone(); // v5 与 v4 两个 crate 各自定义 ClientError，统一归一化为 String。 let result: Result<(), String> = match &client { ClientKind::V5(client) => { @@ -318,7 +323,7 @@ impl MqttEngine { engine.emit(EngineEvent::Published { connection_id, topic: params.topic, - payload: params.payload, + payload: sent_payload, qos: params.qos, retain: params.retain, content_type: params.content_type, @@ -490,6 +495,22 @@ impl MqttEngine { rx } + /// 驻留字符串（主题/连接 ID）：同一内容全进程只保留一份 `Arc`。 + /// 缓存有上限，异常流量下整体清空重建，避免无限增长。 + pub(crate) fn intern(&self, s: &str) -> Arc { + const INTERN_CAP: usize = 10_000; + let mut cache = self.topics.lock().unwrap(); + if let Some(arc) = cache.get(s) { + return arc.clone(); + } + if cache.len() >= INTERN_CAP { + cache.clear(); + } + let arc: Arc = Arc::from(s); + cache.insert(s.to_string(), arc.clone()); + arc + } + pub fn test_connection( self: &Arc, cfg: ConnectionConfig, @@ -942,14 +963,15 @@ async fn test_handshake(cfg_in: &ConnectionConfig) -> Result<(), String> { // ─── 事件循环 ──────────────────────────────────────────────────────────────── -fn v5_record(seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord { +fn v5_record(engine: &MqttEngine, seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord { let props = p.properties.as_ref(); + let (payload, payload_truncated) = crate::model::retained_payload(&p.payload); MqttRecord { seq, - connection_id: conn_id.to_string(), + connection_id: engine.intern(conn_id), direction: Direction::Received, - topic: String::from_utf8_lossy(&p.topic).to_string(), - payload: String::from_utf8_lossy(&p.payload).to_string(), + topic: engine.intern(&String::from_utf8_lossy(&p.topic)), + payload, qos: v5_qos_u8(p.qos), retain: p.retain, timestamp: now_ms(), @@ -966,6 +988,7 @@ fn v5_record(seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord { message_expiry_interval: props.and_then(|p| p.message_expiry_interval), subscription_identifier: props .and_then(|p| p.subscription_identifiers.first().map(|v| *v as u32)), + payload_truncated, } } @@ -997,7 +1020,7 @@ fn spawn_v5_loop( match event { Ok(V5Event::Incoming(V5Packet::Publish(p))) => { let topic = String::from_utf8_lossy(&p.topic).to_string(); - let record = v5_record(engine.next_seq(), &id, &p); + let record = v5_record(&engine, engine.next_seq(), &id, &p); engine.log( &id, LogLevel::Info, @@ -1158,12 +1181,14 @@ fn spawn_v4_loop( event = eventloop.poll() => { match event { Ok(rv4::Event::Incoming(rv4::Packet::Publish(p))) => { + let (payload, payload_truncated) = + crate::model::retained_payload(&p.payload); let record = MqttRecord { seq: engine.next_seq(), - connection_id: id.clone(), + connection_id: engine.intern(&id), direction: Direction::Received, - topic: String::from_utf8_lossy(&p.topic).to_string(), - payload: String::from_utf8_lossy(&p.payload).to_string(), + topic: engine.intern(&String::from_utf8_lossy(&p.topic)), + payload, qos: match p.qos { rv4::QoS::AtMostOnce => 0, rv4::QoS::AtLeastOnce => 1, @@ -1177,6 +1202,7 @@ fn spawn_v4_loop( correlation_data: None, message_expiry_interval: None, subscription_identifier: None, + payload_truncated, }; engine.log( &id, LogLevel::Info, "publish_received", diff --git a/gpui-app/src/ui/app.rs b/gpui-app/src/ui/app.rs index 7757d29..69b5847 100644 --- a/gpui-app/src/ui/app.rs +++ b/gpui-app/src/ui/app.rs @@ -2,6 +2,7 @@ use std::collections::{HashMap, VecDeque}; use std::path::PathBuf; +use std::sync::Arc; use gpui_kit::component::button::{Button, ButtonVariants as _, ButtonVariant}; use gpui_kit::component::scroll::ScrollableElement as _; @@ -24,7 +25,7 @@ use crate::aliyun::AliyunPreset; use crate::model::{ AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry, LogLevel, MqttRecord, PublishParams, PublishPreset, SubscribeOptions, Subscription, - ThemeModePref, + ThemeModePref, retained_payload, }; use crate::mqtt::{EngineEvent, MqttEngine}; use crate::store::Storage; @@ -83,7 +84,8 @@ pub struct MqttXApp { // ── 运行时状态 ── pub statuses: HashMap, pub errors: HashMap, - pub messages: HashMap>, + /// 每连接消息环形缓冲：键为驻留后的连接 ID，容量受 max_messages 约束 + pub messages: HashMap, VecDeque>, pub logs: VecDeque, pub seq: u64, @@ -308,7 +310,7 @@ impl MqttXApp { let id = record.connection_id.clone(); record.seq = self.next_seq(); self.push_message(id.clone(), record); - if let Some(view) = self.views.get(&id) { + if let Some(view) = self.views.get(id.as_ref()) { view.update(cx, |_, cx| cx.notify()); } } @@ -321,11 +323,12 @@ impl MqttXApp { content_type, user_properties, } => { + let (payload, payload_truncated) = retained_payload(&payload); let record = MqttRecord { seq: self.next_seq(), - connection_id: connection_id.clone(), + connection_id: Arc::from(connection_id.as_str()), direction: Direction::Published, - topic, + topic: Arc::from(topic.as_str()), payload, qos, retain, @@ -336,9 +339,10 @@ impl MqttXApp { correlation_data: None, message_expiry_interval: None, subscription_identifier: None, + payload_truncated, }; - self.push_message(connection_id.clone(), record); - if let Some(view) = self.views.get(&connection_id) { + self.push_message(Arc::from(connection_id.as_str()), record); + if let Some(view) = self.views.get(connection_id.as_str()) { view.update(cx, |_, cx| cx.notify()); } } @@ -400,14 +404,17 @@ impl MqttXApp { self.seq } - fn push_message(&mut self, connection_id: String, record: MqttRecord) { + fn push_message(&mut self, connection_id: Arc, record: MqttRecord) { let cap = self.settings.max_messages.max(100); - let list = self.messages.entry(connection_id).or_default(); - if list.len() >= cap { - let drop_n = list.len() - cap + 1; - list.drain(0..drop_n); + let list = 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(record); + list.push_back(record); } fn push_log(&mut self, entry: LogEntry) { diff --git a/gpui-app/src/ui/connection_view.rs b/gpui-app/src/ui/connection_view.rs index 884849a..7e9494e 100644 --- a/gpui-app/src/ui/connection_view.rs +++ b/gpui-app/src/ui/connection_view.rs @@ -24,8 +24,9 @@ use gpui_kit::{ use gpui_kit::component::IndexPath; use crate::model::{ - render_template, render_will_templates, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, - LogLevel, MqttRecord, PayloadFormat, PublishParams, SubscribeOptions, Subscription, + render_template, render_will_templates, ConnectionConfig, ConnectionStatus, Direction, + GlobalVariable, LogLevel, MqttRecord, PayloadFormat, PublishParams, SubscribeOptions, + Subscription, MAX_PAYLOAD_RETAIN, }; use crate::mqtt::MqttEngine; use crate::ui::app::MqttXApp; @@ -1285,7 +1286,7 @@ impl ConnectionView { .font_family(mono.clone()) // 匹配到订阅色时主题文字着色，否则保持默认前景色 .when_some(sub_color, |t, hue| t.text_color(hue_color(hue))) - .child(record.topic.clone()), + .child(record.topic.to_string()), ) .child( div() @@ -1315,6 +1316,17 @@ impl ConnectionView { .child("retain"), ) }) + .when(record.payload_truncated, |t| { + t.child( + div() + .text_xs() + .px_1() + .rounded_sm() + .bg(cx.theme().warning.alpha(0.15)) + .text_color(cx.theme().warning) + .child(format!("已截断 {}KB", MAX_PAYLOAD_RETAIN / 1024)), + ) + }) .child( div() .text_xs() @@ -1385,7 +1397,7 @@ impl ConnectionView { .tooltip("复制主题") .on_click(move |_, window, cx| { cx.stop_propagation(); - cx.write_to_clipboard(ClipboardItem::new_string(copy_topic.clone())); + cx.write_to_clipboard(ClipboardItem::new_string(copy_topic.to_string())); window.push_notification(Notification::success("已复制主题"), cx); }), ) @@ -1398,7 +1410,7 @@ impl ConnectionView { .tooltip("复制负载原文") .on_click(move |_, window, cx| { cx.stop_propagation(); - cx.write_to_clipboard(ClipboardItem::new_string(copy_payload.clone())); + cx.write_to_clipboard(ClipboardItem::new_string(copy_payload.to_string())); window.push_notification(Notification::success("已复制负载"), cx); }), ) @@ -1533,7 +1545,7 @@ impl ConnectionView { .collect(); let mut matched = 0usize; let mut records: Vec = Vec::new(); - if let Some(m) = app.messages.get(&conn) { + if let Some(m) = app.messages.get(conn.as_str()) { for r in m.iter().rev() { // 方向 / 搜索 / 订阅过滤叠加，过滤后再截断保证最新消息优先 let dir_ok = match dir { @@ -1742,7 +1754,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); + app.messages.remove(clear_id.as_str()); cx.notify(); }); } diff --git a/gpui-app/tests/engine_live.rs b/gpui-app/tests/engine_live.rs index f950f01..7d42119 100644 --- a/gpui-app/tests/engine_live.rs +++ b/gpui-app/tests/engine_live.rs @@ -108,7 +108,7 @@ fn wait_subscribe(rx: &smol::channel::Receiver, topic: &str) -> boo fn wait_message(rx: &smol::channel::Receiver, topic: &str) -> bool { pump(rx, Duration::from_secs(10), |ev| match ev { - EngineEvent::Message(m) => m.topic == topic && m.payload == "hello-gpui", + EngineEvent::Message(m) => &*m.topic == topic && &*m.payload == "hello-gpui", _ => false, }) }