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`。
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 = "Vec::is_empty")]
51    pub degraded: Vec<Degrade>,
52}
53
54impl LogEvent {
55    /// 新事件，时间戳取当前本地时间。
56    pub fn new(kind: &str) -> Self {
57        Self {
58            ts: Local::now().to_rfc3339_opts(SecondsFormat::Millis, false),
59            kind: kind.to_string(),
60            ms: 0,
61            stages: BTreeMap::new(),
62            candidates: None,
63            folded: None,
64            rerank_docs: None,
65            rerank_tokens: None,
66            hits: None,
67            degraded: Vec::new(),
68        }
69    }
70}
71
72/// 进程内的事件接收位。检索线程取出 `Arc` 后立刻放掉这把锁。
73#[derive(Default)]
74pub(crate) struct EventRegistry {
75    sink: Mutex<Option<EventSink>>,
76}
77
78impl EventRegistry {
79    pub fn get(&self) -> Option<EventSink> {
80        self.sink.lock().clone()
81    }
82    pub fn set(&self, sink: EventSink) {
83        *self.sink.lock() = Some(sink);
84    }
85    pub fn clear(&self) -> bool {
86        self.sink.lock().take().is_some()
87    }
88    pub fn is_registered(&self) -> bool {
89        self.sink.lock().is_some()
90    }
91}
92
93/// 阶段计时：每 `mark` 一次，记下距上一次 `mark` 的毫秒数。
94pub(crate) struct StageTimer {
95    last: Instant,
96    marks: BTreeMap<String, u64>,
97}
98
99impl StageTimer {
100    pub fn start() -> Self {
101        Self { last: Instant::now(), marks: BTreeMap::new() }
102    }
103    pub fn mark(&mut self, name: &str) {
104        let now = Instant::now();
105        self.marks.insert(name.to_string(), (now - self.last).as_millis() as u64);
106        self.last = now;
107    }
108    /// 收尾：把最后一段也记进去，返回整张表。
109    pub fn finish(mut self, name: &str) -> BTreeMap<String, u64> {
110        self.mark(name);
111        self.marks
112    }
113}
p_memory/events.rs

p_memory/
events.rs
