perf: 消息内存按字节预算,杜绝大报文流撑爆内存
- 新增 MessageRing:条数上限 + 每连接 32MB 字节预算双重驱逐 (从最旧丢弃,均摊 O(1));预分配容量按设置钳制在 512 - 单条 payload 保留上限 256KB → 128KB;MqttRecord::retained_bytes 计入文本与原始字节,供预算驱逐与资源监控展示 - 清空消息统一走 clear_messages;设置页标注字节预算兜底说明 - 新增单元测试:条数驱逐、字节预算、超预算单条保留、驱逐顺序
This commit is contained in:
+139
-1
@@ -6,6 +6,8 @@ use std::sync::Arc;
|
|||||||
|
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
use std::collections::VecDeque;
|
||||||
|
|
||||||
// ─── 连接配置 ────────────────────────────────────────────────────────────────
|
// ─── 连接配置 ────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
/// MQTT 协议版本。
|
/// MQTT 协议版本。
|
||||||
@@ -451,9 +453,70 @@ pub struct MqttRecord {
|
|||||||
pub raw_bytes: Option<Arc<[u8]>>,
|
pub raw_bytes: Option<Arc<[u8]>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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<MqttRecord>,
|
||||||
|
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 的保留上限。超限截断,避免个别大报文
|
/// 消息历史中单条 payload 的保留上限。超限截断,避免个别大报文
|
||||||
/// 长期占据历史缓冲(MQTTX 亦有类似限制)。
|
/// 长期占据历史缓冲(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 字节转成用于历史记录的共享文本,超限截断。
|
/// 把原始 payload 字节转成用于历史记录的共享文本,超限截断。
|
||||||
/// 返回 (文本, 非文本时的原始字节, 是否截断)。截断时回退到 UTF-8 字符边界。
|
/// 返回 (文本, 非文本时的原始字节, 是否截断)。截断时回退到 UTF-8 字符边界。
|
||||||
@@ -985,4 +1048,79 @@ mod tests {
|
|||||||
b.ssl.ca_file = "/tmp/ca.pem".into();
|
b.ssl.ca_file = "/tmp/ca.pem".into();
|
||||||
assert!(a.session_params_changed(&b), "TLS 配置改动应触发重连提示");
|
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));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+14
-14
@@ -24,8 +24,8 @@ use gpui_kit::{
|
|||||||
use crate::aliyun::AliyunPreset;
|
use crate::aliyun::AliyunPreset;
|
||||||
use crate::model::{
|
use crate::model::{
|
||||||
AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry,
|
AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry,
|
||||||
LogLevel, MqttRecord, PublishParams, PublishPreset, SubscribeOptions, Subscription,
|
LogLevel, MqttRecord, MessageRing, PublishParams, PublishPreset, SubscribeOptions,
|
||||||
ThemeModePref, retained_payload,
|
Subscription, ThemeModePref, retained_payload,
|
||||||
};
|
};
|
||||||
use crate::mqtt::{EngineEvent, MqttEngine};
|
use crate::mqtt::{EngineEvent, MqttEngine};
|
||||||
use crate::store::Storage;
|
use crate::store::Storage;
|
||||||
@@ -84,8 +84,8 @@ pub struct MqttXApp {
|
|||||||
// ── 运行时状态 ──
|
// ── 运行时状态 ──
|
||||||
pub statuses: HashMap<String, ConnectionStatus>,
|
pub statuses: HashMap<String, ConnectionStatus>,
|
||||||
pub errors: HashMap<String, String>,
|
pub errors: HashMap<String, String>,
|
||||||
/// 每连接消息环形缓冲:键为驻留后的连接 ID,容量受 max_messages 约束
|
/// 每连接消息环形缓冲:键为驻留后的连接 ID,条数 + 字节预算双重约束
|
||||||
pub messages: HashMap<Arc<str>, VecDeque<MqttRecord>>,
|
pub messages: HashMap<Arc<str>, MessageRing>,
|
||||||
pub logs: VecDeque<LogEntry>,
|
pub logs: VecDeque<LogEntry>,
|
||||||
pub seq: u64,
|
pub seq: u64,
|
||||||
|
|
||||||
@@ -473,16 +473,16 @@ impl MqttXApp {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn push_message(&mut self, connection_id: Arc<str>, record: MqttRecord) {
|
fn push_message(&mut self, connection_id: Arc<str>, record: MqttRecord) {
|
||||||
let cap = self.settings.max_messages.max(100);
|
let cap = self.settings.max_messages;
|
||||||
let list = self
|
self.messages
|
||||||
.messages
|
|
||||||
.entry(connection_id)
|
.entry(connection_id)
|
||||||
.or_insert_with(|| VecDeque::with_capacity(cap));
|
.or_insert_with(|| MessageRing::new(cap))
|
||||||
// 环形缓冲:满则丢最旧,均摊 O(1)(原来的 Vec::drain 是 O(n) 搬移)
|
.push(record, cap);
|
||||||
while list.len() >= cap {
|
}
|
||||||
list.pop_front();
|
|
||||||
}
|
/// 清空指定连接的消息(内存缓冲随 drop 释放)。
|
||||||
list.push_back(record);
|
pub fn clear_messages(&mut self, id: &str) {
|
||||||
|
self.messages.remove(id);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn push_log(&mut self, entry: LogEntry) {
|
fn push_log(&mut self, entry: LogEntry) {
|
||||||
@@ -603,7 +603,7 @@ impl MqttXApp {
|
|||||||
self.engine.close(id, false);
|
self.engine.close(id, false);
|
||||||
self.connections.retain(|c| c.id != id);
|
self.connections.retain(|c| c.id != id);
|
||||||
self.subscriptions.retain(|s| s.connection_id != id);
|
self.subscriptions.retain(|s| s.connection_id != id);
|
||||||
self.messages.remove(id);
|
self.clear_messages(id);
|
||||||
self.statuses.remove(id);
|
self.statuses.remove(id);
|
||||||
self.errors.remove(id);
|
self.errors.remove(id);
|
||||||
self.views.remove(id);
|
self.views.remove(id);
|
||||||
|
|||||||
@@ -1765,7 +1765,7 @@ impl ConnectionView {
|
|||||||
view.detail_format = DetailFormat::Auto;
|
view.detail_format = DetailFormat::Auto;
|
||||||
if let Some(app) = view.app.upgrade() {
|
if let Some(app) = view.app.upgrade() {
|
||||||
app.update(cx, |app, cx| {
|
app.update(cx, |app, cx| {
|
||||||
app.messages.remove(clear_id.as_str());
|
app.clear_messages(&clear_id);
|
||||||
cx.notify();
|
cx.notify();
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ use gpui_kit::{
|
|||||||
Styled as _, Window,
|
Styled as _, Window,
|
||||||
};
|
};
|
||||||
|
|
||||||
use crate::model::{AppSettings, ThemeModePref};
|
use crate::model::{AppSettings, ThemeModePref, MAX_CONNECTION_MESSAGE_BYTES};
|
||||||
use crate::mqtt::MqttEngine;
|
use crate::mqtt::MqttEngine;
|
||||||
use crate::ui::app::MqttXApp;
|
use crate::ui::app::MqttXApp;
|
||||||
use crate::ui::widgets::{field, make_select, OptionDelegate};
|
use crate::ui::widgets::{field, make_select, OptionDelegate};
|
||||||
@@ -388,6 +388,15 @@ impl Render for SettingsDialog {
|
|||||||
"每条连接内存中保留的消息条数(100~100000)",
|
"每条连接内存中保留的消息条数(100~100000)",
|
||||||
Input::new(&self.max_messages),
|
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(
|
.child(
|
||||||
h_flex()
|
h_flex()
|
||||||
.justify_between()
|
.justify_between()
|
||||||
|
|||||||
Reference in New Issue
Block a user