p_memory/
events.rs

1//! 库内事件流：把「这次执行发生了什么、各阶段花了多久」交给宿主。
2//!
3//! 库只产出事件，不决定去向——落盘、轮转、保留多久都是宿主日志体系的事。
4//! 未注册 sink 时全程不构造事件、不格式化，开销为零。
5
6use crate::types::Degrade;
7use chrono::{Local, SecondsFormat};
8use parking_lot::Mutex;
9use serde::{Deserialize, Serialize};
10use std::collections::BTreeMap;
11use std::sync::Arc;
12use std::time::Instant;
13
14/// 事件接收回调。库在检索线程里同步调用它，因此它必须非阻塞：
15/// 在 sink 里做同步 IO 或网络上报，会把检索拖住，和慢的重排回调一样。
16pub type EventSink = Arc<dyn Fn(&LogEvent) + Send + Sync>;
17
18/// 一条事件。`kind` 决定哪些字段有值，无值的字段不出现。
19///
20/// 事件里不含查询原文与正文：宿主本来就知道查询词，库再抄一遍只是把检索词
21/// 散进宿主的日志文件。
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct LogEvent {
24    /// 本地时间的 RFC3339 时间戳，毫秒精度。
25    pub ts: String,
26    /// 事件类型：`search`、`index_rebuild`。
27    pub kind: String,
28    /// 本次执行的总耗时（毫秒）。
29    pub ms: u64,
30    /// 各阶段耗时（毫秒）。`rerank` 是「等宿主重排回调返回」的时间，也就是模型
31    /// 推理时间，不是库的开销；`embed` 同理。
32    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
33    pub stages: BTreeMap<String, u64>,
34    /// 参与融合的候选条数。
35    #[serde(default, skip_serializing_if = "Option::is_none")]
36    pub candidates: Option<usize>,
37    /// 折叠后剩下的条数。
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub folded: Option<usize>,
40    /// 实际送进重排回调的文档数。
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub rerank_docs: Option<usize>,
43    /// 实际送进重排回调的文档 token 数（含查询词）。
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub rerank_tokens: Option<usize>,
46    /// 最终返回的命中条数。
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub hits: Option<usize>,
49    /// 重建写入索引的文档数。
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub documents: Option<usize>,
52    /// 索引格式串。
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub format: Option<String>,
55    /// 本次落在哪几档降级。
56    #[serde(default, skip_serializing_if = "Vec::is_empty")]
57    pub degraded: Vec<Degrade>,
58}
59
60impl LogEvent {
61    /// 新事件，时间戳取当前本地时间。
62    pub fn new(kind: &str) -> Self {
63        Self {
64            ts: Local::now().to_rfc3339_opts(SecondsFormat::Millis, false),
65            kind: kind.to_string(),
66            ms: 0,
67            stages: BTreeMap::new(),
68            candidates: None,
69            folded: None,
70            rerank_docs: None,
71            rerank_tokens: None,
72            hits: None,
73            documents: None,
74            format: None,
75            degraded: Vec::new(),
76        }
77    }
78}
79
80/// 进程内的事件接收位。检索线程取出 `Arc` 后立刻放掉这把锁。
81#[derive(Default)]
82pub(crate) struct EventRegistry {
83    sink: Mutex<Option<EventSink>>,
84}
85
86impl EventRegistry {
87    pub fn get(&self) -> Option<EventSink> {
88        self.sink.lock().clone()
89    }
90    pub fn set(&self, sink: EventSink) {
91        *self.sink.lock() = Some(sink);
92    }
93    pub fn clear(&self) -> bool {
94        self.sink.lock().take().is_some()
95    }
96    pub fn is_registered(&self) -> bool {
97        self.sink.lock().is_some()
98    }
99}
100
101/// 阶段计时：每 `mark` 一次，记下距上一次 `mark` 的毫秒数。
102pub(crate) struct StageTimer {
103    last: Instant,
104    marks: BTreeMap<String, u64>,
105}
106
107impl StageTimer {
108    pub fn start() -> Self {
109        Self { last: Instant::now(), marks: BTreeMap::new() }
110    }
111    pub fn mark(&mut self, name: &str) {
112        let now = Instant::now();
113        self.marks.insert(name.to_string(), (now - self.last).as_millis() as u64);
114        self.last = now;
115    }
116    /// 收尾：把最后一段也记进去，返回整张表。
117    pub fn finish(mut self, name: &str) -> BTreeMap<String, u64> {
118        self.mark(name);
119        self.marks
120    }
121}
p_memory/events.rs

p_memory/
events.rs
