lmflow/ffi/graph/
input_poller.rs

1#![allow(clippy::missing_safety_doc)]
2use super::*;
3
4#[no_mangle]
5pub unsafe extern "C" fn lmflow_graph_input(
6    g: *mut LMFlowGraph,
7    port: *const c_char,
8) -> *mut LMFlowInput {
9    guard_val(std::ptr::null_mut(), || {
10        let Some(slot) = slot_mut(g) else {
11            last_error::set("graph handle is null");
12            return std::ptr::null_mut();
13        };
14        let Some(gr) = slot.graph.as_ref() else {
15            last_error::set("graph not yet initialized");
16            return std::ptr::null_mut();
17        };
18        let Some(name) = cstr(port) else {
19            last_error::set("port name is null");
20            return std::ptr::null_mut();
21        };
22        let inner = gr.inner().clone();
23        match inner.input_edge_by_name(name) {
24            Some(edge) => {
25                // 调用方拥有:独立 Box,持一份 Arc<GraphInner>。须 lmflow_input_free 释放。
26                Box::into_raw(Box::new(InputHandle { graph: inner, edge })) as *mut LMFlowInput
27            }
28            None => {
29                last_error::set(&format!("graph input port `{name}` does not exist"));
30                std::ptr::null_mut()
31            }
32        }
33    })
34}
35
36unsafe fn input_ref<'a>(i: *mut LMFlowInput) -> Option<&'a InputHandle> {
37    if i.is_null() {
38        None
39    } else {
40        Some(&*(i as *const InputHandle))
41    }
42}
43
44#[no_mangle]
45pub unsafe extern "C" fn lmflow_input_send(i: *mut LMFlowInput, pkt: LMFlowPacket) -> i32 {
46    guard(|| {
47        let p = take_packet(pkt);
48        match input_ref(i) {
49            Some(h) => to_status(h.graph.send_by_edge(h.edge, p, true)),
50            None => fail(Error::InvalidArg("input handle is null".into())),
51        }
52    })
53}
54
55#[no_mangle]
56pub unsafe extern "C" fn lmflow_input_try_send(i: *mut LMFlowInput, pkt: LMFlowPacket) -> i32 {
57    guard(|| {
58        let p = take_packet(pkt);
59        match input_ref(i) {
60            Some(h) => to_status(h.graph.send_by_edge(h.edge, p, false)),
61            None => fail(Error::InvalidArg("input handle is null".into())),
62        }
63    })
64}
65
66#[no_mangle]
67pub unsafe extern "C" fn lmflow_input_close(i: *mut LMFlowInput) {
68    guard_val((), || {
69        if let Some(h) = input_ref(i) {
70            h.graph.close_edge_pub(h.edge);
71        }
72    });
73}
74
75#[no_mangle]
76pub unsafe extern "C" fn lmflow_input_free(i: *mut LMFlowInput) {
77    guard_val((), || {
78        if !i.is_null() {
79            // 调用方拥有:归还这份句柄(及其对引擎的 Arc)。图可能已 free,但句柄仍安全。
80            drop(Box::from_raw(i as *mut InputHandle));
81        }
82    });
83}
84
85#[no_mangle]
86pub unsafe extern "C" fn lmflow_graph_add_packet(
87    g: *mut LMFlowGraph,
88    port: *const c_char,
89    pkt: LMFlowPacket,
90) -> i32 {
91    guard(|| {
92        let p = take_packet(pkt);
93        let Some(name) = cstr(port) else {
94            return fail(Error::InvalidArg("port name is null".into()));
95        };
96        with_graph(g, |gr| {
97            let inner = gr.inner();
98            match inner.input_edge_by_name(name) {
99                Some(e) => to_status(inner.send_by_edge(e, p, true)),
100                None => fail(Error::NotFound(format!(
101                    "graph input port `{name}` does not exist"
102                ))),
103            }
104        })
105    })
106}
107
108#[no_mangle]
109pub unsafe extern "C" fn lmflow_graph_close_input(g: *mut LMFlowGraph, port: *const c_char) -> i32 {
110    guard(|| {
111        let Some(name) = cstr(port) else {
112            return fail(Error::InvalidArg("port name is null".into()));
113        };
114        with_graph(g, |gr| to_status(gr.close_input(name)))
115    })
116}
117
118#[no_mangle]
119pub unsafe extern "C" fn lmflow_graph_close_all_inputs(g: *mut LMFlowGraph) {
120    guard_val((), || {
121        let _ = with_graph(g, |gr| {
122            gr.close_all_inputs();
123            code::OK
124        });
125    });
126}
127
128#[no_mangle]
129pub unsafe extern "C" fn lmflow_graph_add_poller(
130    g: *mut LMFlowGraph,
131    port: *const c_char,
132) -> *mut LMFlowPoller {
133    lmflow_graph_add_poller_ex(g, port, false)
134}
135
136#[no_mangle]
137pub unsafe extern "C" fn lmflow_graph_add_poller_ex(
138    g: *mut LMFlowGraph,
139    port: *const c_char,
140    observe_timestamp_bounds: bool,
141) -> *mut LMFlowPoller {
142    guard_val(std::ptr::null_mut(), || {
143        let Some(slot) = slot_mut(g) else {
144            last_error::set("graph handle is null");
145            return std::ptr::null_mut();
146        };
147        let Some(gr) = slot.graph.as_ref() else {
148            last_error::set("graph not yet initialized");
149            return std::ptr::null_mut();
150        };
151        let Some(name) = cstr(port) else {
152            last_error::set("port name is null");
153            return std::ptr::null_mut();
154        };
155        let poller = if observe_timestamp_bounds {
156            gr.add_poller_with_timestamp_bounds(name)
157        } else {
158            gr.add_poller(name)
159        };
160        match poller {
161            Ok(p) => {
162                // 调用方拥有:独立 Box,持一份 Arc<GraphInner>。须 lmflow_poller_free 释放。
163                Box::into_raw(Box::new(p)) as *mut LMFlowPoller
164            }
165            Err(e) => {
166                last_error::set(&e.to_string());
167                std::ptr::null_mut()
168            }
169        }
170    })
171}
172
173#[no_mangle]
174pub unsafe extern "C" fn lmflow_graph_add_poller_bounded(
175    g: *mut LMFlowGraph,
176    port: *const c_char,
177    capacity: usize,
178    overflow: i32,
179) -> *mut LMFlowPoller {
180    guard_val(std::ptr::null_mut(), || {
181        let Some(slot) = slot_mut(g) else {
182            last_error::set("graph handle is null");
183            return std::ptr::null_mut();
184        };
185        let Some(gr) = slot.graph.as_ref() else {
186            last_error::set("graph not yet initialized");
187            return std::ptr::null_mut();
188        };
189        let Some(name) = cstr(port) else {
190            last_error::set("port name is null");
191            return std::ptr::null_mut();
192        };
193        let overflow = match overflow {
194            0 => crate::graph::PollerOverflow::Block,
195            1 => crate::graph::PollerOverflow::DropOldest,
196            2 => crate::graph::PollerOverflow::DropNewest,
197            3 => crate::graph::PollerOverflow::Latest,
198            _ => {
199                last_error::set(
200                    "unknown poller overflow policy (valid: BLOCK / DROP_OLDEST / DROP_NEWEST / LATEST)",
201                );
202                return std::ptr::null_mut();
203            }
204        };
205        match gr.add_poller_with_options(name, crate::graph::PollerOptions::new(capacity, overflow))
206        {
207            Ok(poller) => Box::into_raw(Box::new(poller)) as *mut LMFlowPoller,
208            Err(error) => {
209                last_error::set(&error.to_string());
210                std::ptr::null_mut()
211            }
212        }
213    })
214}
215
216unsafe fn poller_ref<'a>(p: *mut LMFlowPoller) -> Option<&'a Poller> {
217    if p.is_null() {
218        None
219    } else {
220        Some(&*(p as *const Poller))
221    }
222}
223
224#[no_mangle]
225pub unsafe extern "C" fn lmflow_poller_dropped_count(p: *mut LMFlowPoller) -> u64 {
226    guard_val(0, || poller_ref(p).map_or(0, Poller::dropped_count))
227}
228
229#[no_mangle]
230pub unsafe extern "C" fn lmflow_poller_next_status(
231    p: *mut LMFlowPoller,
232    out: *mut LMFlowPacket,
233) -> i32 {
234    guard(|| {
235        let Some(poller) = poller_ref(p) else {
236            return fail(Error::InvalidArg("poller handle is null".into()));
237        };
238        match poller.next_result() {
239            Ok(Some(pkt)) => {
240                if !out.is_null() {
241                    *out = own_packet(pkt);
242                }
243                code::OK
244            }
245            Ok(None) => code::CLOSED,
246            Err(error) => fail(error),
247        }
248    })
249}
250
251#[no_mangle]
252pub unsafe extern "C" fn lmflow_poller_try_next_status(
253    p: *mut LMFlowPoller,
254    out: *mut LMFlowPacket,
255) -> i32 {
256    guard(|| {
257        let Some(poller) = poller_ref(p) else {
258            return fail(Error::InvalidArg("poller handle is null".into()));
259        };
260        match poller.try_next_result() {
261            Ok(Some(pkt)) => {
262                if !out.is_null() {
263                    *out = own_packet(pkt);
264                }
265                code::OK
266            }
267            Ok(None) => {
268                // `Ok(None)` 这里可能来自 `try_next_result` 的**未关闭**分支 —— 即「读那一刻
269                // 队列空且未 closed」。下面这次 `is_closed()` 是**更晚的另一次 load**:生产者
270                // 可以在两次之间入队并关边,此时只看 `is_closed()` 就会在**队列非空**时报
271                // `CLOSED`,而 C 侧据此停止排空 —— 与 `Poller::next` 里刚修掉的丢包是同一个
272                // 竞争,只是搬到了 ABI 边界上。
273                //
274                // 因此必须两个条件都成立才算流结束。顺序不能反:先观察到 `closed == true`
275                // 就意味着不会再有入队,随后的 `is_empty()` 才是权威的;反过来先看空、再看
276                // closed,中间又是同一个窗口。队列非空时返回 `WOULD_BLOCK` 让调用方再来一次,
277                // 下一次的 `pop` 就会取到那些包。
278                if poller.is_closed() && poller.is_empty() {
279                    code::CLOSED
280                } else {
281                    code::WOULD_BLOCK
282                }
283            }
284            Err(error) => fail(error),
285        }
286    })
287}
288
289#[no_mangle]
290pub unsafe extern "C" fn lmflow_poller_next_timeout(
291    p: *mut LMFlowPoller,
292    out: *mut LMFlowPacket,
293    timeout_ms: i64,
294) -> i32 {
295    guard(|| {
296        let Some(poller) = poller_ref(p) else {
297            return fail(Error::InvalidArg("poller handle is null".into()));
298        };
299        match poller.next_timeout(std::time::Duration::from_millis(timeout_ms.max(0) as u64)) {
300            Ok(Some(pkt)) => {
301                if !out.is_null() {
302                    *out = own_packet(pkt);
303                }
304                code::OK
305            }
306            Ok(None) => code::CLOSED,
307            Err(e) => fail(e),
308        }
309    })
310}
311
312#[no_mangle]
313pub unsafe extern "C" fn lmflow_poller_free(p: *mut LMFlowPoller) {
314    guard_val((), || {
315        if !p.is_null() {
316            // 调用方拥有:归还这份句柄(及其对引擎的 Arc)。图可能已 free,但句柄仍安全。
317            drop(Box::from_raw(p as *mut Poller));
318        }
319    });
320}
lmflow/ffi/graph/input_poller.rs

lmflow/ffi/graph/
input_poller.rs
