lmflow 0.1.1

Dataflow-graph compute engine: a pure-Rust scheduler behind a stable C ABI
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156

//! 引擎自带默认 Rust 算子(`src/builtin.rs`)的端到端验收 —— **纯 Rust,零 C++**。
//!
//! 这些测试在两种 feature 配置下都跑(不带 `#![cfg]`),因为默认算子任何配置下都在。
//! 只有两个算子,且都是纯结构性的:`PassThrough`(接线)与 `Sink`(自行终结分支)。
//! 最后一条钉住「扇出是引擎原生能力」—— 正因如此才不需要 Split 算子。

use std::time::Duration;

use lmflow::{Graph, Packet, State, Timestamp};

#[test]
fn passthrough_forwards_unchanged() {
    let g = Graph::from_yaml(
        r#"
nodes:
  - { name: a, kernel: PassThrough, input_ports: ["in"], output_ports: ["mid"] }
  - { name: b, kernel: PassThrough, input_ports: ["mid"], output_ports: ["out"] }
input_ports: ["in"]
output_ports: ["out"]
"#,
    )
    .unwrap();
    let out = g.add_poller("out").unwrap();
    g.start().unwrap();
    let inp = g.input("in").unwrap();
    for i in 0..5i64 {
        inp.send(Packet::from_i64(i * 10).at(Timestamp(i))).unwrap();
    }
    g.close_all_inputs();
    let mut got = Vec::new();
    while got.len() < 5 {
        match out.next() {
            Some(p) => got.push(p.as_i64().unwrap()),
            None => break,
        }
    }
    assert_eq!(got, vec![0, 10, 20, 30, 40], "两级 PassThrough 应原样透传");
    g.wait_done_timeout(Duration::from_secs(5)).unwrap();
    assert_eq!(g.state(), State::Terminated);
}

#[test]
fn sink_consumes_and_counts() {
    // Sink 零输出口 —— 用**按图**计数器断言它真的收到了包并正常收尾。
    let g = Graph::from_yaml(
        r#"
nodes:
  - { name: k, kernel: Sink, input_ports: ["in"], output_ports: [] }
input_ports: ["in"]
output_ports: []
"#,
    )
    .unwrap();
    g.start().unwrap();
    let inp = g.input("in").unwrap();
    for i in 0..3i64 {
        inp.send(Packet::from_i64(i).at(Timestamp(i))).unwrap();
    }
    g.close_all_inputs();
    g.wait_done_timeout(Duration::from_secs(5)).unwrap();
    assert_eq!(g.counter_value("sink.packets"), 3);
    assert_eq!(g.counter_value("sink.closed"), 1);
    assert_eq!(g.state(), State::Terminated);
}

/// 两个默认算子都注册了。
#[test]
fn defaults_are_registered() {
    lmflow::builtin::register_defaults();
    let names = lmflow::kernel::registered_names();
    for want in ["PassThrough", "Sink"] {
        assert!(names.iter().any(|n| n == want), "{want} 应已注册:{names:?}");
    }
}

/// **扇出是引擎原生能力**:一条边可直接挂多个消费者,无需任何 Split 算子。
/// 这条是「默认算子里不放 Split」这个决定的依据。
#[test]
fn fanout_is_native_no_split_kernel_needed() {
    let g = Graph::from_yaml(
        r#"
nodes:
  - { name: a, kernel: PassThrough, input_ports: ["in"],  output_ports: ["mid"] }
  - { name: b, kernel: PassThrough, input_ports: ["mid"], output_ports: ["x"] }
  - { name: c, kernel: PassThrough, input_ports: ["mid"], output_ports: ["y"] }
input_ports: ["in"]
output_ports: ["x", "y"]
"#,
    )
    .unwrap();
    let px = g.add_poller("x").unwrap();
    let py = g.add_poller("y").unwrap();
    g.start().unwrap();
    g.input("in")
        .unwrap()
        .send(Packet::from_i64(7).at(Timestamp(0)))
        .unwrap();
    g.close_all_inputs();
    assert_eq!(px.next().unwrap().as_i64(), Some(7));
    assert_eq!(
        py.next().unwrap().as_i64(),
        Some(7),
        "同一条边的两个消费者都应收到"
    );
    g.wait_done_timeout(Duration::from_secs(5)).unwrap();
}

/// 日志快路:默认没装 sink 时 `log_enabled()` 为假(算子据此跳过 `format!`),
/// 装上之后必须为真且真的收到日志 —— 快路不能把日志吞掉。
#[test]
fn log_fast_path_does_not_swallow_logs() {
    use std::ffi::{c_char, c_void, CStr};
    use std::sync::atomic::{AtomicUsize, Ordering};

    static SEEN: AtomicUsize = AtomicUsize::new(0);
    unsafe extern "C" fn sink_cb(_u: *mut c_void, _lvl: i32, msg: *const c_char) {
        if !msg.is_null() && !unsafe { CStr::from_ptr(msg) }.to_bytes().is_empty() {
            SEEN.fetch_add(1, Ordering::Relaxed);
        }
    }

    // 默认:无 sink → 快路生效
    assert!(
        !lmflow::runtime::log_enabled(),
        "默认不该有 log sink(快路应生效,算子可跳过 format!)"
    );

    lmflow::ffi::lmflow_set_log_callback(Some(sink_cb), std::ptr::null_mut());
    assert!(lmflow::runtime::log_enabled(), "装上 sink 后快路须让路");

    // Sink 算子每包一条 debug 日志 + close 一条 info —— 装了 sink 就该真收到
    let g = Graph::from_yaml(
        r#"
nodes:
  - { name: k, kernel: Sink, input_ports: ["in"], output_ports: [] }
input_ports: ["in"]
output_ports: []
"#,
    )
    .unwrap();
    g.start().unwrap();
    let inp = g.input("in").unwrap();
    for i in 0..3i64 {
        inp.send(Packet::from_i64(i).at(Timestamp(i))).unwrap();
    }
    g.close_all_inputs();
    g.wait_done_timeout(Duration::from_secs(5)).unwrap();

    lmflow::ffi::lmflow_set_log_callback(None, std::ptr::null_mut());
    assert!(!lmflow::runtime::log_enabled(), "卸掉 sink 后应回到快路");
    assert!(
        SEEN.load(Ordering::Relaxed) >= 3,
        "装了 sink 就必须真收到日志(每包一条 + close 一条),实际 {}",
        SEEN.load(Ordering::Relaxed)
    );
}