lmflow/
kernel.rs

1//! 算子:vtable 抽象、全局注册表、端口表与契约。
2
3use std::collections::BTreeMap;
4use std::ffi::c_void;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::{LazyLock, Mutex};
7
8use crate::config::{parse_port_spec, RouteConfig, RouteMode};
9use crate::context::Context;
10use crate::status::{Error, Result};
11
12/// 算子的函数指针表(布局与 `include/flow.h` 的 `LMFlowKernelVTable` 一致)。
13/// `ctx`/`contract` 参数在 C 侧是 `LMFlowContext*`/`LMFlowContract*`,此处用 `*mut c_void`
14/// 以免模块间循环依赖;ffi 层负责转换。
15#[repr(C)]
16#[derive(Clone, Copy)]
17pub struct KernelVTable {
18    pub create: Option<unsafe extern "C" fn(*mut c_void) -> *mut c_void>,
19    pub get_contract: Option<unsafe extern "C" fn(*mut c_void, *mut c_void)>,
20    pub open: Option<unsafe extern "C" fn(*mut c_void, *mut c_void) -> i32>,
21    pub process: Option<unsafe extern "C" fn(*mut c_void, *mut c_void) -> i32>,
22    pub close: Option<unsafe extern "C" fn(*mut c_void, *mut c_void) -> i32>,
23    pub destroy: Option<unsafe extern "C" fn(*mut c_void)>,
24}
25
26#[derive(Clone, Copy)]
27pub struct KernelReg {
28    pub vtable: KernelVTable,
29    pub factory: *mut c_void,
30    pub language: KernelLanguage,
31}
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34#[repr(u32)]
35pub enum KernelLanguage {
36    Unknown = 0,
37    Rust = 1,
38    Cpp = 2,
39    Python = 3,
40    C = 4,
41}
42
43impl KernelLanguage {
44    pub fn label(self) -> &'static str {
45        match self {
46            Self::Unknown => "unknown",
47            Self::Rust => "Rust",
48            Self::Cpp => "C++",
49            Self::Python => "Python",
50            Self::C => "C",
51        }
52    }
53}
54
55// 安全性:factory 由注册方提供、要求指向静态存储或长生命周期对象;
56// 注册表只在 init 阶段读取,不做解引用以外的操作。
57unsafe impl Send for KernelReg {}
58unsafe impl Sync for KernelReg {}
59
60/// 全局算子注册表。用 BTreeMap 以保证枚举顺序确定(便于诊断输出可复现)。
61static REGISTRY: LazyLock<Mutex<BTreeMap<String, KernelReg>>> =
62    LazyLock::new(|| Mutex::new(BTreeMap::new()));
63
64pub fn register(name: &str, vtable: KernelVTable, factory: *mut c_void) -> Result<()> {
65    register_with_language(name, vtable, factory, KernelLanguage::Unknown)
66}
67
68pub fn register_with_language(
69    name: &str,
70    vtable: KernelVTable,
71    factory: *mut c_void,
72    language: KernelLanguage,
73) -> Result<()> {
74    if name.is_empty() {
75        return Err(Error::InvalidArg("kernel name must not be empty".into()));
76    }
77    if vtable.process.is_none() {
78        return Err(Error::InvalidArg(format!(
79            "kernel `{name}` vtable is missing process (the only required callback)"
80        )));
81    }
82    let mut reg = REGISTRY.lock().expect("registry lock poisoned");
83    if reg.contains_key(name) {
84        return Err(Error::InvalidArg(format!(
85            "kernel `{name}` already registered"
86        )));
87    }
88    reg.insert(
89        name.to_string(),
90        KernelReg {
91            vtable,
92            factory,
93            language,
94        },
95    );
96    Ok(())
97}
98
99pub fn lookup(name: &str) -> Option<KernelReg> {
100    REGISTRY
101        .lock()
102        .expect("registry lock poisoned")
103        .get(name)
104        .copied()
105}
106
107pub fn registered_names() -> Vec<String> {
108    REGISTRY
109        .lock()
110        .expect("registry lock poisoned")
111        .keys()
112        .cloned()
113        .collect()
114}
115
116/// 「算子未注册」的统一报错 —— 必须列出可用名字,否则用户无从下手
117/// (常见原因是对应 kernel 组件尚未注册，或算子名拼错)。
118fn not_registered(name: &str) -> Error {
119    let registered = registered_names();
120    let suggestion = crate::diagnostic::did_you_mean(name, registered.iter().map(String::as_str));
121    Error::NotFound(format!(
122        "kernel `{name}` not registered{suggestion}. registered: [{}]",
123        registered.join(", ")
124    ))
125}
126
127/// 一个节点持有的算子实例。Drop 时回调 C 侧 `destroy`(RAII)。
128pub struct KernelInstance {
129    self_ptr: *mut c_void,
130    vtable: KernelVTable,
131    kernel_name: String,
132    language: KernelLanguage,
133    internal: Option<RouteRuntime>,
134}
135
136struct RouteRuleStats {
137    evaluated: AtomicU64,
138    matched: AtomicU64,
139    emitted: AtomicU64,
140}
141
142struct RouteRuntime {
143    config: RouteConfig,
144    rules: Vec<RouteRuleStats>,
145    default_emitted: AtomicU64,
146    unmatched: AtomicU64,
147    dropped: AtomicU64,
148    errors: AtomicU64,
149    missing_metadata: AtomicU64,
150    evaluation_errors: AtomicU64,
151}
152
153#[derive(Debug, Clone)]
154pub(crate) struct RouteStatsSnapshot {
155    pub config: RouteConfig,
156    pub rules: Vec<(u64, u64, u64)>,
157    pub default_emitted: u64,
158    pub unmatched: u64,
159    pub dropped: u64,
160    pub errors: u64,
161    pub missing_metadata: u64,
162    pub evaluation_errors: u64,
163}
164
165// 安全性:self_ptr 指向算子实例,由「节点独占令牌」保证任一时刻只被单线程访问
166// (docs/design.md §7.0 R3)。
167unsafe impl Send for KernelInstance {}
168unsafe impl Sync for KernelInstance {}
169
170impl KernelInstance {
171    pub fn create(kernel_name: &str) -> Result<Self> {
172        let reg = lookup(kernel_name).ok_or_else(|| not_registered(kernel_name))?;
173        let self_ptr = match reg.vtable.create {
174            // 安全性:create 由算子侧提供,契约是返回可传给其它回调的实例指针。
175            Some(f) => unsafe { f(reg.factory) },
176            None => std::ptr::null_mut(), // 无状态算子
177        };
178        Ok(Self {
179            self_ptr,
180            vtable: reg.vtable,
181            kernel_name: kernel_name.to_string(),
182            language: reg.language,
183            internal: None,
184        })
185    }
186
187    pub fn create_route(config: RouteConfig) -> Self {
188        let rules = config
189            .routes
190            .iter()
191            .map(|_| RouteRuleStats {
192                evaluated: AtomicU64::new(0),
193                matched: AtomicU64::new(0),
194                emitted: AtomicU64::new(0),
195            })
196            .collect();
197        Self {
198            self_ptr: std::ptr::null_mut(),
199            vtable: KernelVTable {
200                create: None,
201                get_contract: None,
202                open: None,
203                process: None,
204                close: None,
205                destroy: None,
206            },
207            kernel_name: "__lmflow.route".into(),
208            language: KernelLanguage::Rust,
209            internal: Some(RouteRuntime {
210                config,
211                rules,
212                default_emitted: AtomicU64::new(0),
213                unmatched: AtomicU64::new(0),
214                dropped: AtomicU64::new(0),
215                errors: AtomicU64::new(0),
216                missing_metadata: AtomicU64::new(0),
217                evaluation_errors: AtomicU64::new(0),
218            }),
219        }
220    }
221
222    pub(crate) fn route_stats(&self) -> Option<RouteStatsSnapshot> {
223        let route = self.internal.as_ref()?;
224        Some(RouteStatsSnapshot {
225            config: route.config.clone(),
226            rules: route
227                .rules
228                .iter()
229                .map(|stats| {
230                    (
231                        stats.evaluated.load(Ordering::Relaxed),
232                        stats.matched.load(Ordering::Relaxed),
233                        stats.emitted.load(Ordering::Relaxed),
234                    )
235                })
236                .collect(),
237            default_emitted: route.default_emitted.load(Ordering::Relaxed),
238            unmatched: route.unmatched.load(Ordering::Relaxed),
239            dropped: route.dropped.load(Ordering::Relaxed),
240            errors: route.errors.load(Ordering::Relaxed),
241            missing_metadata: route.missing_metadata.load(Ordering::Relaxed),
242            evaluation_errors: route.evaluation_errors.load(Ordering::Relaxed),
243        })
244    }
245
246    pub(crate) fn reset_stats(&self) {
247        let Some(route) = &self.internal else { return };
248        for stats in &route.rules {
249            stats.evaluated.store(0, Ordering::Relaxed);
250            stats.matched.store(0, Ordering::Relaxed);
251            stats.emitted.store(0, Ordering::Relaxed);
252        }
253        route.default_emitted.store(0, Ordering::Relaxed);
254        route.unmatched.store(0, Ordering::Relaxed);
255        route.dropped.store(0, Ordering::Relaxed);
256        route.errors.store(0, Ordering::Relaxed);
257        route.missing_metadata.store(0, Ordering::Relaxed);
258        route.evaluation_errors.store(0, Ordering::Relaxed);
259    }
260
261    pub fn kernel_name(&self) -> &str {
262        &self.kernel_name
263    }
264
265    pub fn language(&self) -> KernelLanguage {
266        self.language
267    }
268
269    /// 调 C 侧 `get_contract` 填充契约。
270    ///
271    /// # Safety
272    /// `contract` 必须是指向存活 [`Contract`] 的有效指针,且在本调用期间不被其它代码访问。
273    pub unsafe fn fill_contract(kernel_name: &str, contract: *mut c_void) -> Result<()> {
274        let reg = lookup(kernel_name).ok_or_else(|| not_registered(kernel_name))?;
275        if let Some(f) = reg.vtable.get_contract {
276            // 安全性:contract 指向存活的 Contract;回调期内不会被别处访问。
277            unsafe { f(reg.factory, contract) };
278        }
279        Ok(())
280    }
281
282    /// 以下三者传入的 `ctx` 是 `*mut Context` 的擦除形式。
283    /// 调用期间**不持有任何引擎锁**(docs/design.md §7.0 R1)。
284    ///
285    /// # Safety
286    /// `ctx` 必须是指向存活 [`crate::context::Context`] 的有效指针,且调用方持有
287    /// 本节点的**独占令牌**(docs/design.md §7.0 R3)—— 否则算子回调可能与引擎
288    /// 对同一 Context 的访问竞争。
289    pub unsafe fn open(&self, ctx: *mut c_void) -> i32 {
290        let status = match self.vtable.open {
291            Some(f) => unsafe { f(self.self_ptr, ctx) },
292            None => 0,
293        };
294        unsafe { &mut *(ctx as *mut Context) }.callback_status(status)
295    }
296    /// # Safety
297    /// 同 [`Self::open`]。
298    pub unsafe fn process(&self, ctx: *mut c_void) -> i32 {
299        if let Some(route) = &self.internal {
300            let context = unsafe { &mut *(ctx as *mut Context) };
301            return match route_process(route, context) {
302                Ok(()) => 0,
303                Err(error) => {
304                    context.set_error(&error.to_string());
305                    -1
306                }
307            };
308        }
309        let status = match self.vtable.process {
310            Some(f) => unsafe { f(self.self_ptr, ctx) },
311            None => 0,
312        };
313        unsafe { &mut *(ctx as *mut Context) }.callback_status(status)
314    }
315    /// # Safety
316    /// 同 [`Self::open`]。
317    pub unsafe fn close(&self, ctx: *mut c_void) -> i32 {
318        let status = match self.vtable.close {
319            Some(f) => unsafe { f(self.self_ptr, ctx) },
320            None => 0,
321        };
322        unsafe { &mut *(ctx as *mut Context) }.callback_status(status)
323    }
324}
325
326fn route_process(route: &RouteRuntime, context: &mut Context) -> Result<()> {
327    let packet = context.input(0).cloned().unwrap_or_default();
328    let mut matched = false;
329    let mut default_output = None;
330    for (index, rule) in route.config.routes.iter().enumerate() {
331        if rule.default {
332            default_output = Some(rule.to.as_str());
333            continue;
334        }
335        route.rules[index].evaluated.fetch_add(1, Ordering::Relaxed);
336        let predicate = rule
337            .when
338            .as_ref()
339            .ok_or_else(|| Error::InvalidArg("route rule missing condition".into()))?;
340        route.missing_metadata.fetch_add(
341            predicate.missing_metadata_count(packet.metadata()),
342            Ordering::Relaxed,
343        );
344        let hit = match predicate.evaluate(packet.metadata(), packet.timestamp()) {
345            Ok(hit) => hit,
346            Err(error) => {
347                route.evaluation_errors.fetch_add(1, Ordering::Relaxed);
348                return Err(error);
349            }
350        };
351        if hit {
352            route.rules[index].matched.fetch_add(1, Ordering::Relaxed);
353            let out = context
354                .out_ports
355                .index_by_name(&rule.to)
356                .ok_or_else(|| Error::InvalidArg(format!("unknown route output `{}`", rule.to)))?;
357            context.emit(out, packet.clone())?;
358            route.rules[index].emitted.fetch_add(1, Ordering::Relaxed);
359            context.shared.counter_add(
360                &format!("route.{}.matched.{}", context.node_name, rule.to),
361                1,
362            );
363            context.shared.counter_add(
364                &format!("route.{}.emitted.{}", context.node_name, rule.to),
365                1,
366            );
367            matched = true;
368            if route.config.mode == RouteMode::First {
369                break;
370            }
371        }
372    }
373    if !matched {
374        if let Some(port) = default_output {
375            let out = context
376                .out_ports
377                .index_by_name(port)
378                .ok_or_else(|| Error::InvalidArg(format!("unknown default output `{port}`")))?;
379            context.emit(out, packet)?;
380            route.default_emitted.fetch_add(1, Ordering::Relaxed);
381            return Ok(());
382        }
383        route.unmatched.fetch_add(1, Ordering::Relaxed);
384        match route.config.unmatched.as_str() {
385            "drop" => {
386                route.dropped.fetch_add(1, Ordering::Relaxed);
387                context
388                    .shared
389                    .counter_add(&format!("route.{}.dropped", context.node_name), 1);
390                Ok(())
391            }
392            "error" => {
393                route.errors.fetch_add(1, Ordering::Relaxed);
394                Err(Error::Kernel("route packet did not match any rule".into()))
395            }
396            port => {
397                let out = context.out_ports.index_by_name(port).ok_or_else(|| {
398                    Error::InvalidArg(format!("unknown unmatched output `{port}`"))
399                })?;
400                context
401                    .shared
402                    .counter_add(&format!("route.{}.emitted.{}", context.node_name, port), 1);
403                context.emit(out, packet)?;
404                Ok(())
405            }
406        }
407    } else {
408        Ok(())
409    }
410}
411
412impl std::fmt::Debug for KernelInstance {
413    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
414        write!(f, "KernelInstance{{`{}`}}", self.kernel_name)
415    }
416}
417
418impl Drop for KernelInstance {
419    fn drop(&mut self) {
420        if let Some(f) = self.vtable.destroy {
421            if !self.self_ptr.is_null() {
422                unsafe { f(self.self_ptr) };
423            }
424        }
425        self.self_ptr = std::ptr::null_mut();
426    }
427}
428
429/// 端口表:把 YAML 的端口声明整理成「序号 ↔ 名字 ↔ (tag,index)」三向查询。
430///
431/// **扁平序号 = 声明顺序**(ADR #17)。
432#[derive(Debug, Default, Clone)]
433pub struct PortTable {
434    names: Vec<String>,
435    tags: Vec<(String, usize)>,
436    by_name: BTreeMap<String, usize>,
437    by_tag: BTreeMap<(String, usize), usize>,
438}
439
440impl PortTable {
441    pub fn build(decls: &[String], what: &str) -> Result<Self> {
442        let mut t = PortTable::default();
443        // 每个 tag 下自动编号的游标
444        let mut auto: BTreeMap<String, usize> = BTreeMap::new();
445        for decl in decls {
446            let spec = parse_port_spec(decl)?;
447            let idx = match spec.index {
448                Some(i) => i,
449                None => {
450                    let c = auto.entry(spec.tag.clone()).or_insert(0);
451                    let i = *c;
452                    *c += 1;
453                    i
454                }
455            };
456            let flat = t.names.len();
457            let key = (spec.tag.clone(), idx);
458            if t.by_tag.contains_key(&key) {
459                return Err(Error::InvalidArg(format!(
460                    "{what}: port `{decl}` (tag=`{}`, index={idx}) is a duplicate",
461                    spec.tag
462                )));
463            }
464            if t.by_name.contains_key(&spec.name) {
465                return Err(Error::InvalidArg(format!(
466                    "{what}: port name `{}` is duplicated within the same kernel",
467                    spec.name
468                )));
469            }
470            t.by_tag.insert(key, flat);
471            t.by_name.insert(spec.name.clone(), flat);
472            t.names.push(spec.name);
473            t.tags.push((spec.tag, idx));
474        }
475        t.check_tag_index_continuity(what)?;
476        Ok(t)
477    }
478
479    /// 同一 tag 下的 index 必须从 0 连续(否则 `InputId(tag, i)` 会出现空洞)。
480    fn check_tag_index_continuity(&self, what: &str) -> Result<()> {
481        let mut per_tag: BTreeMap<&str, Vec<usize>> = BTreeMap::new();
482        for (tag, idx) in &self.tags {
483            per_tag.entry(tag.as_str()).or_default().push(*idx);
484        }
485        for (tag, mut idxs) in per_tag {
486            idxs.sort_unstable();
487            for (expect, got) in idxs.iter().enumerate() {
488                if expect != *got {
489                    return Err(Error::InvalidArg(format!(
490                        "{what}: tag `{tag}` index must be contiguous from 0, got {idxs:?}"
491                    )));
492                }
493            }
494        }
495        Ok(())
496    }
497
498    pub fn len(&self) -> usize {
499        self.names.len()
500    }
501    pub fn is_empty(&self) -> bool {
502        self.names.is_empty()
503    }
504    pub fn name(&self, flat: usize) -> Option<&str> {
505        self.names.get(flat).map(|s| s.as_str())
506    }
507    pub fn names(&self) -> &[String] {
508        &self.names
509    }
510    pub fn id_by_tag(&self, tag: &str, index: usize) -> Option<usize> {
511        self.by_tag.get(&(tag.to_string(), index)).copied()
512    }
513    pub fn index_by_name(&self, name: &str) -> Option<usize> {
514        self.by_name.get(name).copied()
515    }
516}
517
518/// A kernel's port contract: payload-type constraints plus required side packets.
519///
520/// Port counts and names come from YAML and are already known when the contract is built; the
521/// kernel only fills in the types.
522#[derive(Debug)]
523pub struct Contract {
524    pub inputs: std::sync::Arc<PortTable>,
525    pub outputs: std::sync::Arc<PortTable>,
526    /// 0 = 未声明(接受任意类型)
527    pub input_types: Vec<u64>,
528    pub output_types: Vec<u64>,
529    pub required_side_packets: Vec<String>,
530    error: Option<String>,
531}
532
533impl Contract {
534    pub fn new(inputs: std::sync::Arc<PortTable>, outputs: std::sync::Arc<PortTable>) -> Self {
535        let ni = inputs.len();
536        let no = outputs.len();
537        Self {
538            inputs,
539            outputs,
540            input_types: vec![0; ni],
541            output_types: vec![0; no],
542            required_side_packets: Vec::new(),
543            error: None,
544        }
545    }
546
547    pub(crate) fn record_error(&mut self, message: impl Into<String>) {
548        if self.error.is_none() {
549            self.error = Some(message.into());
550        }
551    }
552
553    pub(crate) fn take_error(&mut self) -> Option<String> {
554        self.error.take()
555    }
556}
557
558#[cfg(test)]
559mod tests {
560    use super::*;
561
562    fn decls(v: &[&str]) -> Vec<String> {
563        v.iter().map(|s| s.to_string()).collect()
564    }
565
566    #[test]
567    fn flat_index_follows_declaration_order() {
568        // ADR #17:扁平序号就是 YAML 里的书写顺序,不按 tag 排序
569        let t = PortTable::build(&decls(&["ZZZ:z", "plain", "AAA:a"]), "test").unwrap();
570        assert_eq!(t.name(0), Some("z"));
571        assert_eq!(t.name(1), Some("plain"));
572        assert_eq!(t.name(2), Some("a"));
573    }
574
575    #[test]
576    fn lookup_by_tag_and_by_name() {
577        let t = PortTable::build(
578            &decls(&["frames", "VIDEO:cam0", "MASK:0:m0", "MASK:1:m1"]),
579            "test",
580        )
581        .unwrap();
582        assert_eq!(t.id_by_tag("VIDEO", 0), Some(1));
583        assert_eq!(t.id_by_tag("MASK", 1), Some(3));
584        assert_eq!(t.id_by_tag("MASK", 5), None);
585        assert_eq!(t.index_by_name("frames"), Some(0));
586        assert_eq!(t.index_by_name("nope"), None);
587        // 无 tag 的归入空 tag,自动编号
588        assert_eq!(t.id_by_tag("", 0), Some(0));
589    }
590
591    #[test]
592    fn auto_index_per_tag() {
593        let t = PortTable::build(&decls(&["A:x", "A:y", "B:z"]), "test").unwrap();
594        assert_eq!(t.id_by_tag("A", 0), Some(0));
595        assert_eq!(t.id_by_tag("A", 1), Some(1));
596        assert_eq!(t.id_by_tag("B", 0), Some(2));
597    }
598
599    #[test]
600    fn rejects_duplicate_and_discontinuous() {
601        assert!(
602            PortTable::build(&decls(&["A:0:x", "A:0:y"]), "t").is_err(),
603            "same tag and index duplicated"
604        );
605        assert!(
606            PortTable::build(&decls(&["a", "a"]), "t").is_err(),
607            "duplicate port name"
608        );
609        let err = PortTable::build(&decls(&["A:1:x"]), "t").unwrap_err();
610        assert!(
611            err.to_string().contains("contiguous"),
612            "index has a gap: {err}"
613        );
614    }
615
616    #[test]
617    fn register_rejects_vtable_without_process() {
618        let vt = KernelVTable {
619            create: None,
620            get_contract: None,
621            open: None,
622            process: None,
623            close: None,
624            destroy: None,
625        };
626        let err = register("__no_process__", vt, std::ptr::null_mut()).unwrap_err();
627        assert!(err.to_string().contains("process"), "{err}");
628    }
629
630    #[test]
631    fn lookup_missing_kernel_lists_available() {
632        let err = KernelInstance::create("__definitely_missing__").unwrap_err();
633        // 报错要能指导用户 —— 列出可用名字
634        assert!(err.to_string().contains("registered"), "{err}");
635    }
636}
lmflow/kernel.rs

lmflow/
kernel.rs
