lmflow/
runtime.rs

1//! 运行期共享设施:日志接收器、图级共享状态、C 字符串驻留。
2
3use std::collections::BTreeMap;
4use std::ffi::{c_void, CString};
5use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
6use std::sync::Mutex;
7
8use crate::buffer_pool::BufferPool;
9use crate::config::GraphConfig;
10use crate::status::Error;
11
12// ---------------------------------------------------------------- 日志
13
14pub const LOG_ERROR: i32 = 0;
15pub const LOG_WARN: i32 = 1;
16pub const LOG_INFO: i32 = 2;
17pub const LOG_DEBUG: i32 = 3;
18
19type LogFn = unsafe extern "C" fn(*mut c_void, i32, *const std::ffi::c_char);
20
21struct LogSink {
22    cb: Option<LogFn>,
23    user: *mut c_void,
24}
25// 安全性:user 是宿主提供的不透明指针,引擎只原样回传。
26unsafe impl Send for LogSink {}
27
28static LOG: Mutex<LogSink> = Mutex::new(LogSink {
29    cb: None,
30    user: std::ptr::null_mut(),
31});
32
33/// 是否装了 sink 的**无锁快路标志**。没装时(默认)`log()` 靠它直接返回,不必去抢
34/// 那把进程级 `LOG` 互斥锁 —— 算子里的 `cc.Log` 可能是**每包**调用的,让默认零配置
35/// 的用户每包抢一次全局锁是不能接受的。仅作快路提示,权威值仍在 `LOG` 里。
36static LOG_SET: AtomicBool = AtomicBool::new(false);
37
38pub fn set_log_callback(cb: Option<LogFn>, user: *mut c_void) {
39    let mut sink = LOG.lock().expect("log lock poisoned");
40    sink.cb = cb;
41    sink.user = user;
42    // 先改 sink 再放标志:置位后 log() 才会去读,故不会读到半成品。
43    LOG_SET.store(cb.is_some(), Ordering::Release);
44}
45
46/// 当前是否装了日志 sink。**给算子做「格式化之前先问一句」用** ——
47/// `format!` 会堆分配,没人听的时候一分钱都不该花:
48///
49/// ```ignore
50/// if lmflow::runtime::log_enabled() {
51///     cc.log(LOG_DEBUG, &format!("frame {}", n));   // 只在真有 sink 时才格式化
52/// }
53/// ```
54pub fn log_enabled() -> bool {
55    LOG_SET.load(Ordering::Acquire)
56}
57
58/// 打一条日志。**调用时不得持有任何引擎内部锁**(见 flow.h 日志一节的承诺):
59/// 回调可能去抢 GIL 或加宿主自己的锁,持锁调用会形成锁序环。
60pub fn log(level: i32, msg: &str) {
61    // 快路:没装 sink 就别碰那把全局锁(见 LOG_SET)。
62    if !LOG_SET.load(Ordering::Acquire) {
63        return;
64    }
65    let (cb, user) = {
66        let sink = LOG.lock().expect("log lock poisoned");
67        (sink.cb, sink.user)
68    };
69    if let Some(f) = cb {
70        if let Ok(c) = CString::new(msg) {
71            unsafe { f(user, level, c.as_ptr()) };
72        }
73    }
74}
75
76pub fn log_warn(msg: &str) {
77    log(LOG_WARN, msg);
78}
79pub fn log_info(msg: &str) {
80    log(LOG_INFO, msg);
81}
82
83// ---------------------------------------------------------------- C 字符串驻留
84//
85// C ABI 里有不少返回 `const char*` 且「生命周期随 graph」的接口。
86// 驻留后指针永久有效:CString 的堆缓冲不随容器搬移而失效。
87
88#[derive(Default)]
89pub struct CStrArena {
90    map: Mutex<BTreeMap<String, CString>>,
91}
92
93impl CStrArena {
94    /// 返回该字符串对应的、生命周期与本 arena 相同的 C 指针。
95    pub fn intern(&self, s: &str) -> *const std::ffi::c_char {
96        let mut m = self.map.lock().expect("string pool lock poisoned");
97        let c = m
98            .entry(s.to_string())
99            .or_insert_with(|| CString::new(s).unwrap_or_default());
100        c.as_ptr()
101    }
102}
103
104/// 线程局部的「最近一次错误」文本,供 `lmflow_last_error` 返回。
105pub mod last_error {
106    use std::cell::RefCell;
107    use std::ffi::{c_char, CString};
108
109    thread_local! {
110        static LAST: RefCell<CString> = RefCell::new(CString::default());
111    }
112
113    pub fn set(msg: &str) {
114        let c = CString::new(msg).unwrap_or_default();
115        LAST.with(|l| *l.borrow_mut() = c);
116    }
117
118    pub fn get() -> *const c_char {
119        LAST.with(|l| l.borrow().as_ptr())
120    }
121
122    pub fn take() -> String {
123        LAST.with(|l| l.borrow().to_string_lossy().into_owned())
124    }
125}
126
127// ---------------------------------------------------------------- 图级共享状态
128
129/// 关闭原因,与 `LMFlowCloseReason` 一致。
130pub const CLOSE_NORMAL: i32 = 0;
131pub const CLOSE_ERROR: i32 = 1;
132pub const CLOSE_CANCELLED: i32 = 2;
133
134pub struct GraphShared {
135    pub config: GraphConfig,
136    pub(crate) buffer_pool: std::sync::Arc<BufferPool>,
137    /// 首个错误(后续错误只记日志,不覆盖首因)
138    error: Mutex<Option<Error>>,
139    error_c: CStrArena,
140    has_error: AtomicBool,
141    cancelled: AtomicBool,
142    /// 全局水位统计
143    total_queued: AtomicUsize,
144    total_queued_bytes: AtomicU64,
145    /// 算子自报计数器
146    counters: Mutex<BTreeMap<String, i64>>,
147    /// 逐次调用 trace 的有界事件环。`config.trace_capacity > 0` 时为 `Some`;`None` 时
148    /// 热路径零开销(scheduler 连时钟都不读)。
149    trace: Option<crate::trace::TraceRing>,
150    pub strings: CStrArena,
151}
152
153impl GraphShared {
154    pub fn new(config: GraphConfig) -> Self {
155        Self {
156            buffer_pool: std::sync::Arc::new(BufferPool::new(config.buffer_pool_max_bytes)),
157            trace: (config.trace_capacity > 0)
158                .then(|| crate::trace::TraceRing::new(config.trace_capacity)),
159            config,
160            error: Mutex::new(None),
161            error_c: CStrArena::default(),
162            has_error: AtomicBool::new(false),
163            cancelled: AtomicBool::new(false),
164            total_queued: AtomicUsize::new(0),
165            total_queued_bytes: AtomicU64::new(0),
166            counters: Mutex::new(BTreeMap::new()),
167            strings: CStrArena::default(),
168        }
169    }
170
171    // ---- 错误 ----
172
173    pub fn record_error(&self, e: Error) {
174        let msg = e.to_string();
175        let mut slot = self.error.lock().expect("error lock poisoned");
176        if slot.is_none() {
177            *slot = Some(e);
178            self.has_error.store(true, Ordering::SeqCst);
179            drop(slot); // 先解锁再打日志:日志回调不得在持锁时调用
180            log(LOG_ERROR, &msg);
181        } else {
182            drop(slot);
183            log(LOG_WARN, &format!("(subsequent error, ignored) {msg}"));
184        }
185    }
186
187    /// 复位所有**运行期**状态,供图 reset 重跑用(`config` 等构建期数据不动)。
188    /// `record_error` 只「首因生效、不覆盖」,没有反向清除路径 —— 这里补上,
189    /// 否则 reset 后的图会带着上一轮的 error / cancelled 出生(start 立刻被挡回)。
190    ///
191    /// 只应在图**静止**时调用(Terminated + is_idle,见 `GraphInner::reset`),故用
192    /// `&self` + 内部可变即可,无并发。
193    pub fn reset_run_state(&self) {
194        *self.error.lock().expect("error lock poisoned") = None;
195        self.has_error.store(false, Ordering::SeqCst);
196        self.cancelled.store(false, Ordering::SeqCst);
197        self.total_queued.store(0, Ordering::Relaxed);
198        self.total_queued_bytes.store(0, Ordering::Relaxed);
199        // 算子自报计数器:新一轮从零计(诊断语义;要跨轮累计的宿主自己在算子里存)。
200        self.counters.lock().expect("counter lock poisoned").clear();
201        // trace 环:新一轮从零记(与计数器同语义)。容量与开关不变。
202        if let Some(trace) = &self.trace {
203            trace.clear();
204        }
205    }
206
207    /// trace 是否开启(`config.trace_capacity > 0`)。热路径据此决定是否读时钟/记 span。
208    #[inline]
209    pub fn trace_enabled(&self) -> bool {
210        self.trace.is_some()
211    }
212
213    /// 记一条 trace span(仅在开启时)。`node` 为节点索引,时间相对图 epoch,单位微秒。
214    #[inline]
215    pub fn record_trace(
216        &self,
217        node: u32,
218        phase: crate::trace::TracePhase,
219        start_us: i64,
220        dur_us: i64,
221        input_ts: i64,
222    ) {
223        if let Some(trace) = &self.trace {
224            trace.record(node, phase, start_us, dur_us, input_ts);
225        }
226    }
227
228    /// 当前 trace 环内容的快照(未开启时为空)。
229    pub fn trace_snapshot(&self) -> Vec<crate::trace::TraceEvent> {
230        self.trace
231            .as_ref()
232            .map(|t| t.snapshot())
233            .unwrap_or_default()
234    }
235
236    pub fn has_error(&self) -> bool {
237        self.has_error.load(Ordering::SeqCst)
238    }
239
240    pub fn first_error(&self) -> Option<Error> {
241        self.error.lock().expect("error lock poisoned").clone()
242    }
243
244    /// 图级错误文本的 C 指针 —— 注意 `lmflow_last_error` 是线程局部的,
245    /// 工作线程上算子的失败原因只能通过这里拿到。
246    pub fn error_cstr(&self) -> *const std::ffi::c_char {
247        let msg = self
248            .first_error()
249            .map(|e| e.to_string())
250            .unwrap_or_default();
251        self.error_c.intern(&msg)
252    }
253
254    // ---- 取消 ----
255
256    pub fn cancel(&self) {
257        self.cancelled.store(true, Ordering::SeqCst);
258    }
259    pub fn is_cancelled(&self) -> bool {
260        self.cancelled.load(Ordering::SeqCst)
261    }
262
263    /// 算子 close 时应告知的原因,供其决定是否提交结果。
264    pub fn close_reason(&self) -> i32 {
265        if self.is_cancelled() {
266            CLOSE_CANCELLED
267        } else if self.has_error() {
268            CLOSE_ERROR
269        } else {
270            CLOSE_NORMAL
271        }
272    }
273
274    // ---- 全局水位 ----
275    //
276    // 这几个是**纯计数器**:除了自己被读回来跟阈值比一下(`over_watermark`),不承载任何
277    // 「发布数据」的语义 —— 没有任何东西经由它们建立 happens-before。所以用 `Relaxed`。
278    //
279    // 这不是抠常数:`on_enqueue`/`on_dequeue` 是**每包每消费者**都要走的,一包就是 4 次
280    // 原子 RMW。x86 上 `lock xadd` 与 ordering 无关,但 **aarch64 上 SeqCst 的 RMW 会生成
281    // 带全屏障的 `ldaxr/stlxr`**,而 Android / iOS arm64 正是发布目标。
282    //
283    // ⚠ 全库其余的 SeqCst 都**审过并有意保留**:终止判定(`input_closed[i] && queue_len==0`
284    // 这种两段式检查)、`has_error` / `cancelled` / `source_done` 的发布、执行器 stop 标志、
285    // 边与 poller 的 closed —— 那些是真同步,降级会坏掉终止正确性。别顺手一起改。
286
287    pub fn on_enqueue(&self, bytes: u64) {
288        self.total_queued.fetch_add(1, Ordering::Relaxed);
289        self.total_queued_bytes.fetch_add(bytes, Ordering::Relaxed);
290    }
291
292    pub fn on_dequeue(&self, bytes: u64) {
293        // 用 saturating 语义,避免任何计数不平衡导致下溢回绕成天文数字
294        let _ = self
295            .total_queued
296            .try_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
297                Some(v.saturating_sub(1))
298            });
299        let _ = self
300            .total_queued_bytes
301            .try_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
302                Some(v.saturating_sub(bytes))
303            });
304    }
305
306    pub fn total_queued(&self) -> usize {
307        self.total_queued.load(Ordering::Relaxed)
308    }
309    pub fn total_queued_bytes(&self) -> u64 {
310        self.total_queued_bytes.load(Ordering::Relaxed)
311    }
312
313    /// 是否已触及全局水位。超限时把压力转化为**图输入口**背压(只在入口刹车不会
314    /// 引入 diamond 死锁,见 docs/design.md §7.5)。
315    ///
316    /// 水位是**软阈值**:读的是 `Relaxed` 快照,故判定可能滞后一两个包 —— 这正是
317    /// 「水位」该有的语义,不需要精确的那一瞬间。
318    pub fn over_watermark(&self) -> bool {
319        let c = &self.config;
320        c.max_queued_packets > 0 && self.total_queued() >= c.max_queued_packets
321    }
322
323    // ---- 计数器 ----
324
325    pub fn counter_add(&self, name: &str, delta: i64) {
326        let mut m = self.counters.lock().expect("counter lock poisoned");
327        *m.entry(name.to_string()).or_insert(0) += delta;
328    }
329    pub fn counter_value(&self, name: &str) -> i64 {
330        self.counters
331            .lock()
332            .expect("counter lock poisoned")
333            .get(name)
334            .copied()
335            .unwrap_or(0)
336    }
337    pub fn counter_names(&self) -> Vec<String> {
338        self.counters
339            .lock()
340            .expect("counter lock poisoned")
341            .keys()
342            .cloned()
343            .collect()
344    }
345}
346
347#[cfg(test)]
348mod tests {
349    use super::*;
350    use crate::config::GraphConfig;
351
352    fn shared() -> GraphShared {
353        GraphShared::new(GraphConfig::from_yaml("nodes: []").unwrap())
354    }
355
356    #[test]
357    fn first_error_wins() {
358        let s = shared();
359        assert!(!s.has_error());
360        s.record_error(Error::Kernel("first".into()));
361        s.record_error(Error::Kernel("second".into()));
362        assert!(s.has_error());
363        // 首因不被覆盖 —— 后续错误往往是首因的连锁反应
364        assert!(s.first_error().unwrap().to_string().contains("first"));
365    }
366
367    #[test]
368    fn close_reason_priority() {
369        let s = shared();
370        assert_eq!(s.close_reason(), CLOSE_NORMAL);
371        s.record_error(Error::Kernel("x".into()));
372        assert_eq!(s.close_reason(), CLOSE_ERROR);
373        s.cancel();
374        assert_eq!(
375            s.close_reason(),
376            CLOSE_CANCELLED,
377            "cancelled takes priority over error"
378        );
379    }
380
381    #[test]
382    fn watermark_counts_packets_and_bytes() {
383        let cfg = GraphConfig::from_yaml("nodes: []\nmax_queued_packets: 2").unwrap();
384        let s = GraphShared::new(cfg);
385        assert!(!s.over_watermark());
386        s.on_enqueue(10);
387        assert!(!s.over_watermark());
388        s.on_enqueue(10);
389        assert!(
390            s.over_watermark(),
391            "reaching the limit counts as over watermark"
392        );
393        s.on_dequeue(10);
394        assert!(!s.over_watermark());
395        assert_eq!(s.total_queued(), 1);
396        assert_eq!(s.total_queued_bytes(), 10);
397    }
398
399    #[test]
400    fn dequeue_never_underflows() {
401        let s = shared();
402        // 即使计数不平衡也不能回绕成天文数字(会让水位判断永久为真)
403        s.on_dequeue(999);
404        assert_eq!(s.total_queued(), 0);
405        assert_eq!(s.total_queued_bytes(), 0);
406    }
407
408    #[test]
409    fn counters_accumulate() {
410        let s = shared();
411        s.counter_add("frames", 1);
412        s.counter_add("frames", 2);
413        assert_eq!(s.counter_value("frames"), 3);
414        assert_eq!(s.counter_value("nope"), 0);
415        assert_eq!(s.counter_names(), vec!["frames".to_string()]);
416    }
417
418    #[test]
419    fn interned_pointers_stay_valid_after_more_inserts() {
420        let a = CStrArena::default();
421        let p1 = a.intern("first");
422        for i in 0..100 {
423            a.intern(&format!("filler{i}"));
424        }
425        let s = unsafe { std::ffi::CStr::from_ptr(p1) };
426        assert_eq!(
427            s.to_str().unwrap(),
428            "first",
429            "interned pointer must stay valid after subsequent inserts"
430        );
431    }
432}
lmflow/runtime.rs

lmflow/
runtime.rs
