lmflow/graph/
node.rs

1use super::*;
2
3pub(super) const LATENCY_BUCKETS: usize = 32;
4
5pub(super) fn latency_bucket_upper_us(bucket: usize) -> u64 {
6    1u64 << bucket.min(LATENCY_BUCKETS - 1)
7}
8
9fn latency_bucket(us: u64) -> usize {
10    if us <= 1 {
11        0
12    } else {
13        (u64::BITS - (us - 1).leading_zeros()) as usize
14    }
15    .min(LATENCY_BUCKETS - 1)
16}
17
18#[derive(Debug)]
19pub(super) struct E2eStats {
20    pub(super) frames: AtomicU64,
21    pub(super) total_us: AtomicU64,
22    pub(super) max_us: AtomicU64,
23    pub(super) buckets: [AtomicU64; LATENCY_BUCKETS],
24}
25
26impl Default for E2eStats {
27    fn default() -> Self {
28        Self {
29            frames: AtomicU64::new(0),
30            total_us: AtomicU64::new(0),
31            max_us: AtomicU64::new(0),
32            buckets: std::array::from_fn(|_| AtomicU64::new(0)),
33        }
34    }
35}
36
37impl E2eStats {
38    pub(super) fn record(&self, us: u64) {
39        self.total_us.fetch_add(us, Ordering::Relaxed);
40        self.max_us.fetch_max(us, Ordering::Relaxed);
41        self.buckets[latency_bucket(us)].fetch_add(1, Ordering::Relaxed);
42        self.frames.fetch_add(1, Ordering::Release);
43    }
44
45    pub(super) fn reset(&self) {
46        self.frames.store(0, Ordering::Relaxed);
47        self.total_us.store(0, Ordering::Relaxed);
48        self.max_us.store(0, Ordering::Relaxed);
49        for bucket in &self.buckets {
50            bucket.store(0, Ordering::Relaxed);
51        }
52    }
53}
54
55/// 输入策略(节点级可插拔,见 docs/design.md §7.10)。
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub enum InputPolicy {
58    /// 所有输入口齐备才触发。默认。
59    Sync,
60    /// 任一输入口有数据即触发,不等其它口。适合无需对齐的旁路处理。
61    Immediate,
62    /// 队列有界:**满则丢弃最旧的包**。实时场景必备 ——
63    /// 摄像头 30fps 而算子只跑 10fps 时,无界队列会让内存无限增长,丢旧帧才是正确取舍。
64    /// 就绪条件同 `Sync`。
65    FixedSize { capacity: usize },
66    /// **部分对齐**:把输入口划成若干组,每组内各自按时间戳对齐;任一组就绪即触发,
67    /// 且本次只带上该组的口(其余口留空)。分组是输入口的一个**完整划分**(每口恰属一组)。
68    /// 用于「A、B 该配对,C 独立」这类图。存的是输入口序号(建图时由端口名解析)。
69    SyncSet { sets: Vec<Vec<usize>> },
70    /// **批处理**:攒够 `size` 个**对齐元组**一次交给算子(`process()` 按批读
71    /// `input_count`/`input_at`);关流时不足一批也刷出。多输入口时按时间戳对齐
72    /// (与 `Sync` 同源),各口取数可以不同。用于批推理 / 窗口聚合。见 docs/design.md §7.10。
73    Batch { size: usize },
74}
75
76impl InputPolicy {
77    /// 从配置构造。`ins` 用于把 `sync_set` 里的端口**名字**解析成序号并校验。
78    pub(super) fn from_config(
79        c: &crate::config::InputPolicyConfig,
80        ins: &crate::kernel::PortTable,
81    ) -> Result<Self> {
82        Ok(match c.r#type.as_str() {
83            "immediate" => InputPolicy::Immediate,
84            "fixed_size" => InputPolicy::FixedSize {
85                capacity: c.capacity.max(1),
86            },
87            "batch" => InputPolicy::Batch {
88                size: c.capacity.max(1),
89            },
90            "sync_set" => {
91                if c.sets.is_empty() {
92                    return Err(Error::InvalidArg(
93                        "sync_set policy must provide sets (input port groups)".into(),
94                    ));
95                }
96                let mut resolved: Vec<Vec<usize>> = Vec::new();
97                let mut seen = vec![false; ins.len()];
98                for set in &c.sets {
99                    if set.is_empty() {
100                        return Err(Error::InvalidArg("sync_set group must not be empty".into()));
101                    }
102                    let mut group = Vec::with_capacity(set.len());
103                    for name in set {
104                        let idx = ins.index_by_name(name).ok_or_else(|| {
105                            Error::InvalidArg(format!(
106                                "sync_set references nonexistent input port `{name}`"
107                            ))
108                        })?;
109                        if seen[idx] {
110                            return Err(Error::InvalidArg(format!(
111                                "sync_set input port `{name}` appears in multiple groups (groups must be disjoint)"
112                            )));
113                        }
114                        seen[idx] = true;
115                        group.push(idx);
116                    }
117                    resolved.push(group);
118                }
119                if let Some(miss) = (0..ins.len()).find(|&i| !seen[i]) {
120                    return Err(Error::InvalidArg(format!(
121                        "sync_set must cover all input ports; at least `{}` is missing (put it in its own group to keep it independent)",
122                        ins.name(miss).unwrap_or("?")
123                    )));
124                }
125                InputPolicy::SyncSet { sets: resolved }
126            }
127            _ => InputPolicy::Sync,
128        })
129    }
130}
131
132/// 一次触发的计划:处理时间戳 + 参与本次的输入口。
133/// `ports = None` 表示「全部口」(Sync / Immediate / FixedSize 的现状,不分配);
134/// `Some(set)` 表示「只这些口」(SyncSet 的就绪组)—— 认领时只对这些口弹包、推进 bound。
135pub(super) struct Ready {
136    pub(super) ts: Timestamp,
137    pub(super) ports: Option<Vec<usize>>,
138    /// 仅 `batch` 策略:就绪判定时已算好的取包计划。放在这里而不是认领时重算,
139    /// 是为了保住「每口只拿一次队列锁」(ADR #36)—— 判定期已把各口时间戳前缀
140    /// 快照过一次,认领期照计划批量弹出即可,不必再逐轮加锁。
141    pub(super) batch: Option<BatchPlan>,
142}
143
144/// `batch` 策略的认领计划:每个**正向口**本次取多少个包,以及本批末尾的对齐时间戳。
145///
146/// 各口取数**可以不同** —— 某口在某个对齐时间戳上没有包,该轮就不取它。这与 `sync`
147/// 单包时的语义一致(`Context::input_count` 本就是按口计数的),而不是「各口各自数够
148/// `size` 个」:后者会把 0 号口的第 k 个与 1 号口的第 k 个配成一对,而它们未必是同一帧,
149/// 属于**静默的错误配对**。
150pub(super) struct BatchPlan {
151    /// (端口号, 取包数)
152    pub(super) take: Vec<(usize, usize)>,
153    /// 本批最后一轮对齐到的时间戳:用作 `input_ts`,并据此推进各口 bound。
154    pub(super) last_ts: Timestamp,
155}
156
157// ---------------------------------------------------------------- 状态机
158
159/// 与 `LMFlowGraphState` 一致。
160#[derive(Debug, Clone, Copy, PartialEq, Eq)]
161pub enum State {
162    Created = 0,
163    Initialized = 1,
164    Running = 2,
165    Draining = 3,
166    Terminated = 4,
167}
168
169/// Graphviz DOT 输出的详细程度。
170#[derive(Debug, Clone, Copy, PartialEq, Eq)]
171pub enum DotView {
172    /// 仅拓扑、执行器与静态容量。
173    Topology,
174    /// 节点状态与核心吞吐/延迟统计,隐藏逐端口诊断和 Poller 细节。
175    Compact,
176    /// 完整运行诊断,包括端口队列、背压、Poller 与诊断图例。
177    Diagnostics,
178}
179
180// ---------------------------------------------------------------- 边
181
182// Serialize an input's committed sends and closure without holding a Mutex
183// while invoking host callbacks. Reentrant sends fail rather than self-deadlock;
184// reentrant close is completed by the active sender after dispatch finishes.
185#[derive(Default)]
186pub(super) struct IngressState {
187    owner: Option<std::thread::ThreadId>,
188    closing: bool,
189}
190
191pub(super) struct IngressGuard<'a>(Option<&'a Edge>);
192impl IngressGuard<'_> {
193    /// Atomically hand off ownership or retain it to commit a pending close.
194    /// A deferred closer must never arrive between the last closing check and
195    /// releasing the sender, otherwise nobody would finish that close.
196    pub(super) fn release_unless_closing(&mut self) -> bool {
197        let edge = self.0.expect("ingress guard already released");
198        let mut state = edge.ingress.lock().expect("ingress lock poisoned");
199        if state.closing {
200            return true;
201        }
202        state.owner = None;
203        self.0 = None;
204        drop(state);
205        edge.ingress_changed.notify_all();
206        false
207    }
208}
209impl Drop for IngressGuard<'_> {
210    fn drop(&mut self) {
211        if let Some(edge) = self.0 {
212            edge.ingress.lock().expect("ingress lock poisoned").owner = None;
213            edge.ingress_changed.notify_all();
214        }
215    }
216}
217
218pub struct Edge {
219    pub name: String,
220    pub producer: Option<NodeId>,
221    /// (消费者节点, 它的第几个输入口)
222    pub consumers: Vec<(NodeId, usize)>,
223    pub is_graph_input: bool,
224    pub is_graph_output: bool,
225    pub(super) closed: AtomicBool,
226    pub(super) dropped: AtomicU64,
227    pub(super) watermark_backpressure: BackpressureStats,
228    /// 该边上最近一次投递的时间戳。**必须独立记录**,不能拿「队列里还剩的包」当参照 ——
229    /// 队列一排空参照就消失了,回退的时间戳就能混进来。
230    pub(super) last_sent: Mutex<Timestamp>,
231    ingress: Mutex<IngressState>,
232    ingress_changed: Condvar,
233    /// 最近一次发布给输出订阅者的时间戳边界。边界事件只允许单调推进，
234    /// 重复/回退值不再次通知。
235    pub(super) last_published_bound: Mutex<Timestamp>,
236    /// 快路标记：默认没有边界订阅时，逐包派发不触碰新增的锁。
237    pub(super) has_timestamp_bound_subscriber: AtomicBool,
238    pub(super) pollers: Mutex<Vec<Arc<PollerInner>>>,
239    pub(super) observers: Mutex<Vec<Observer>>,
240}
241
242impl Edge {
243    pub(super) fn new(name: String) -> Self {
244        Self {
245            name,
246            producer: None,
247            consumers: Vec::new(),
248            is_graph_input: false,
249            is_graph_output: false,
250            closed: AtomicBool::new(false),
251            dropped: AtomicU64::new(0),
252            watermark_backpressure: BackpressureStats::default(),
253            last_sent: Mutex::new(Timestamp::unset()),
254            ingress: Mutex::new(IngressState::default()),
255            ingress_changed: Condvar::new(),
256            last_published_bound: Mutex::new(Timestamp::unstarted()),
257            has_timestamp_bound_subscriber: AtomicBool::new(false),
258            pollers: Mutex::new(Vec::new()),
259            observers: Mutex::new(Vec::new()),
260        }
261    }
262    pub(super) fn is_ingress_owner(&self) -> bool {
263        self.ingress.lock().expect("ingress lock poisoned").owner
264            == Some(std::thread::current().id())
265    }
266
267    pub(super) fn enter_ingress(&self, blocking: bool) -> Result<IngressGuard<'_>> {
268        let current = std::thread::current().id();
269        let mut state = self.ingress.lock().expect("ingress lock poisoned");
270        loop {
271            if state.closing {
272                return Err(Error::Closed);
273            }
274            if state.owner == Some(current) {
275                return Err(Error::State(
276                    "cannot send to the same input from its dispatch callback".into(),
277                ));
278            }
279            if state.owner.is_none() {
280                state.owner = Some(current);
281                return Ok(IngressGuard(Some(self)));
282            }
283            if !blocking {
284                return Err(Error::WouldBlock);
285            }
286            state = self
287                .ingress_changed
288                .wait(state)
289                .expect("ingress lock poisoned");
290        }
291    }
292
293    pub(super) fn enter_ingress_close(&self, defer_if_busy: bool) -> Option<IngressGuard<'_>> {
294        let current = std::thread::current().id();
295        let mut state = self.ingress.lock().expect("ingress lock poisoned");
296        state.closing = true;
297        self.ingress_changed.notify_all();
298        if state.owner == Some(current) || (defer_if_busy && state.owner.is_some()) {
299            return None;
300        }
301        while state.owner.is_some() {
302            state = self
303                .ingress_changed
304                .wait(state)
305                .expect("ingress lock poisoned");
306        }
307        state.owner = Some(current);
308        Some(IngressGuard(Some(self)))
309    }
310
311    pub(super) fn reset_ingress(&self) {
312        *self.ingress.lock().expect("ingress lock poisoned") = IngressState::default();
313    }
314
315    pub fn is_closed(&self) -> bool {
316        self.closed.load(Ordering::SeqCst)
317    }
318    /// 纯诊断计数器(经 `lmflow_graph_dropped_count` 读回),不承载任何 happens-before
319    /// → `Relaxed`。注意上面的 `closed` 是**真同步**(参与终止判定),必须留 SeqCst。
320    pub fn dropped_count(&self) -> u64 {
321        self.dropped.load(Ordering::Relaxed)
322    }
323}
324
325/// 推模式订阅者。C 宿主给函数指针,Rust 宿主给闭包 —— 后者才是 Rust 侧的自然写法。
326#[derive(Clone)]
327pub(super) enum Observer {
328    C {
329        cb: unsafe extern "C" fn(*mut c_void, crate::ffi::LMFlowPacket),
330        user: *mut c_void,
331        observe_timestamp_bounds: bool,
332    },
333    Rust {
334        callback: Arc<dyn Fn(&Packet) + Send + Sync>,
335        observe_timestamp_bounds: bool,
336    },
337}
338// 安全性:user 是宿主的不透明指针,引擎只原样回传。
339unsafe impl Send for Observer {}
340// ---------------------------------------------------------------- 节点
341
342/// 节点调度状态。支持 `max_in_flight` 个并行 in-flight 调用。
343///
344/// 并行模型(docs/design.md §7.11):
345///  * 认领(try_claim)在锁下**原子地**取一个 context 槽、按对齐时间戳弹入本次输入、
346///    分配一个递增序号 `seq`。序号即消费时间戳的顺序(因为总取最小就绪时间戳)。
347///  * 调用完成后按 `seq` **重排刷新**:只有轮到 `next_flush_seq` 时才把输出投递下游,
348///    从而即使后面的时间戳先算完,下游看到的时间戳依然单调。
349///  * `in_flight` = 已认领但尚未刷新的数量 = 占用中的槽数;它归零节点才可关闭。
350///
351/// `max_in_flight == 1` 是自然特例:只有一个槽,序号恒连续,重排是恒等操作 ——
352/// 行为与串行路径一致。
353#[derive(Debug)]
354pub(super) struct NodeSched {
355    pub(super) opened: bool,
356    /// 已「认领关流」—— 在锁下置位,保证并发时只有一个线程会调算子的 Close。
357    /// 必须与 `closed` 分开:`closed` 表示 Close 已跑完,终止判定看它。
358    pub(super) close_started: bool,
359    pub(super) closed: bool,
360    /// 已认领但尚未刷新的调用数(= 占用中的槽数)。归零方可关闭。
361    pub(super) in_flight: usize,
362    /// 可用的 context 槽序号(初始 0..max_in_flight)。
363    pub(super) free_slots: Vec<usize>,
364    /// 已认领、等待某个 worker 来执行的调用:(slot, seq)。
365    pub(super) ready: VecDeque<(usize, u64)>,
366    /// 取时间戳时分配的下一个序号。
367    pub(super) next_seq: u64,
368    /// 下一个可刷新的序号(保证下游时间戳单调)。
369    pub(super) next_flush_seq: u64,
370    /// 完成但等待按序刷新的调用:seq -> (slot, 是否成功)。
371    pub(super) pending_flush: BTreeMap<u64, (usize, bool)>,
372    /// 当前按序轮到、但因下游内部输入队列已满而暂缓刷新的槽。
373    pub(super) blocked_flush: Option<BlockedFlush>,
374    /// 是否已有线程在做刷新 —— 保证刷新按序、串行(否则并发刷新会打乱下游顺序)。
375    pub(super) flushing: bool,
376    /// Source 本次调用刷新完成后，如何延迟唤醒。
377    pub(super) source_reschedule: Option<SourceReschedule>,
378}
379
380#[derive(Debug, Clone, Copy, PartialEq, Eq)]
381pub(super) enum SourceWaitReason {
382    Rate,
383    Yield,
384    RateAndYield,
385}
386
387#[derive(Debug, Clone, Copy)]
388pub(super) struct SourceReschedule {
389    pub(super) delay: std::time::Duration,
390    pub(super) reason: SourceWaitReason,
391}
392impl NodeSched {
393    pub(super) fn new(max_in_flight: usize) -> Self {
394        Self {
395            opened: false,
396            close_started: false,
397            closed: false,
398            in_flight: 0,
399            free_slots: (0..max_in_flight).collect(),
400            ready: VecDeque::new(),
401            next_seq: 0,
402            next_flush_seq: 0,
403            pending_flush: BTreeMap::new(),
404            blocked_flush: None,
405            flushing: false,
406            source_reschedule: None,
407        }
408    }
409}
410
411/// 节点级运行统计。**全原子、无锁** —— 每包每节点都要更新,放 `Mutex` 里就是在热路径上
412/// 加锁(改造前每包要拿 4 次锁:计时进/出 + 耗时 + processed)。
413///
414/// 计数器用 `Relaxed`:它们不参与任何 happens-before 推理,只被读侧当快照看。
415/// `max_in_flight > 1` 时同一节点会被多个工作线程并发更新,故必须是多写者安全的。
416#[derive(Debug)]
417pub(super) struct NodeStats {
418    pub(super) processed: AtomicU64,
419    pub(super) errors: AtomicU64,
420    pub(super) total_us: AtomicI64,
421    pub(super) max_us: AtomicI64,
422    pub(super) latency_buckets: [AtomicU64; LATENCY_BUCKETS],
423    pub(super) cow_copies: AtomicU64,
424    pub(super) cow_bytes: AtomicU64,
425    /// 本节点从输入口取走的包数(在 `try_claim` 弹包处累加)
426    pub(super) packets_in: AtomicU64,
427    /// 本节点产出并派发下游的包数(在 `flush_staging` 派发处累加)
428    pub(super) packets_out: AtomicU64,
429    /// 下游入队时观察到的**队列深度峰值**(高水位)—— 定位积压点
430    pub(super) peak_queue_depth: AtomicUsize,
431    /// 正在执行算子回调的并发数(> 0 即「在跑」)
432    pub(super) in_flight: AtomicUsize,
433    /// 最近一次 `in_flight` 0→1 跃变的时刻(相对 [`GraphInner::epoch`] 的微秒)。
434    /// **归零时不清零** —— 读侧一律先看 `in_flight > 0` 再用它,从而避开
435    /// 「清零」与「新一次开始」互相覆盖的竞争(那会让诊断值瞬时错乱)。
436    pub(super) started_us: AtomicI64,
437}
438
439impl Default for NodeStats {
440    fn default() -> Self {
441        Self {
442            processed: AtomicU64::new(0),
443            errors: AtomicU64::new(0),
444            total_us: AtomicI64::new(0),
445            max_us: AtomicI64::new(0),
446            latency_buckets: std::array::from_fn(|_| AtomicU64::new(0)),
447            cow_copies: AtomicU64::new(0),
448            cow_bytes: AtomicU64::new(0),
449            packets_in: AtomicU64::new(0),
450            packets_out: AtomicU64::new(0),
451            peak_queue_depth: AtomicUsize::new(0),
452            in_flight: AtomicUsize::new(0),
453            started_us: AtomicI64::new(0),
454        }
455    }
456}
457
458impl NodeStats {
459    pub(super) fn record_latency(&self, us: u64) {
460        self.latency_buckets[latency_bucket(us)].fetch_add(1, Ordering::Relaxed);
461    }
462
463    /// 全字段清零,供图 reset 重跑用。仅在图静止时调用(无并发),但字段是内嵌原子,
464    /// 故用 `&self` 逐个 store 即可(不需要 `&mut`)。
465    pub(super) fn reset(&self) {
466        self.processed.store(0, Ordering::Relaxed);
467        self.errors.store(0, Ordering::Relaxed);
468        self.total_us.store(0, Ordering::Relaxed);
469        self.max_us.store(0, Ordering::Relaxed);
470        for bucket in &self.latency_buckets {
471            bucket.store(0, Ordering::Relaxed);
472        }
473        self.cow_copies.store(0, Ordering::Relaxed);
474        self.cow_bytes.store(0, Ordering::Relaxed);
475        self.packets_in.store(0, Ordering::Relaxed);
476        self.packets_out.store(0, Ordering::Relaxed);
477        self.peak_queue_depth.store(0, Ordering::Relaxed);
478        self.in_flight.store(0, Ordering::Relaxed);
479        self.started_us.store(0, Ordering::Relaxed);
480    }
481}
482
483/// 算子失败时本节点怎么办(见 `NodeConfig::on_error`)。
484#[derive(Debug, Clone, Copy, PartialEq, Eq)]
485pub enum OnError {
486    /// 记录首个错误、终止全图(默认,历史行为)。
487    Abort,
488    /// 丢掉出错的那一个包、**推进下游时间戳边界**、计数并打 WARN,然后继续跑。
489    ///
490    /// 为什么必须推进边界:不推进的话下游会永远等这一刻的数据 ——
491    /// 那等于把「一帧出错」升级成「整图卡死」,比 abort 还糟。
492    Skip,
493}
494
495impl OnError {
496    pub(super) fn from_config(s: &str) -> Self {
497        // 未知值已在 config 校验期拒掉,这里只认已知的两个。
498        if s == "skip" {
499            OnError::Skip
500        } else {
501            OnError::Abort
502        }
503    }
504}
505
506pub struct Node {
507    pub name: String,
508    pub kernel_name: String,
509    pub inputs: Vec<EdgeId>,
510    pub outputs: Vec<EdgeId>,
511    pub in_ports: Arc<PortTable>,
512    pub out_ports: Arc<PortTable>,
513    /// `None` = 宿主主线程(默认执行器,ADR #16);`Some(i)` = executors[i] 线程池
514    /// 本节点所属执行器在 `GraphInner::executors` 中的下标。
515    pub(super) executor: usize,
516    pub(super) policy: InputPolicy,
517    pub(super) input_types: Vec<u64>,
518    pub(super) output_types: Vec<u64>,
519    pub(super) kernel: KernelInstance,
520    /// 每次并行 in-flight 调用一个 context 槽(池大小 = max_in_flight)。
521    /// 用 UnsafeCell 而非 Mutex:算子回调期间引擎必须交出一个 `*mut Context` 给 C 侧,
522    /// 若同时持有 Mutex guard 的 `&mut`,回调里从裸指针再造 `&mut` 就构成别名 UB。
523    /// 独占性由「一个槽同一时刻只被一个调用持有」保证(槽在锁下认领/归还)。
524    /// 池在 build 后不再增长,故元素地址稳定 —— 交给 C 侧的 `*mut Context` 在
525    /// 调用期间始终有效。
526    pub(super) ctxs: Vec<UnsafeCell<Context>>,
527    pub(super) max_in_flight: usize,
528    pub(super) sched: Mutex<NodeSched>,
529    /// 全原子,无锁 —— 见 [`NodeStats`]
530    pub(super) stats: NodeStats,
531    /// 每个输入口一条独立队列(见模块头注释)
532    pub(super) input_queues: Vec<Mutex<VecDeque<Packet>>>,
533    /// 每个正向输入口的无损包数容量。`None` = 不限。
534    pub(super) input_queue_capacity: Vec<Option<usize>>,
535    /// 已由上游刷新预留、尚未真正入队的槽数。与 queue len 合计做并发容量判定。
536    pub(super) input_queue_reserved: Vec<AtomicUsize>,
537    /// 当前各输入队列内 payload 的浅字节数。
538    pub(super) input_queue_bytes: Vec<AtomicU64>,
539    /// 每个输入口的背压与高水位统计。
540    pub(super) input_queue_stats: Vec<InputQueueStats>,
541    pub(super) input_closed: Vec<AtomicBool>,
542    /// 算子失败时的处理策略(见 [`OnError`])。建图期定下,之后不变。
543    pub(super) on_error: OnError,
544    /// 源节点定速:相邻两次 `process` 的最小间隔。`None` = 不限速(见 `NodeConfig::rate`)。
545    pub(super) min_period: Option<std::time::Duration>,
546    /// 上次 `process` 的开始时刻,配合 `min_period` 节流。仅源节点用到 ——
547    /// 源本就串行自续产(一个包跑完才排下一个),故一把 Mutex 足够、无竞争压力。
548    pub(super) last_fire: Mutex<Option<Instant>>,
549    /// 每个输入口是否为 back-edge(反馈寄存器):true 的口不参与就绪 / 终止 / 对齐,
550    /// 入队走 cap-1 drop-old(只留最新反馈)。长度恒 = 输入口数(无 back-edge 则全 false)。
551    pub(super) input_is_back_edge: Vec<bool>,
552    /// 源节点(0 输入口)自报「已产完」。置位后 readiness 不再放行、节点可关流终止。
553    pub(super) source_done: AtomicBool,
554    /// Source 正在协作式等待下一次唤醒；等待期间不可再次认领。
555    pub(super) source_waiting: AtomicBool,
556    /// 当前 Source 等待原因。0=无，1=rate，2=source_yield，3=两者共同约束。
557    pub(super) source_wait_reason: AtomicUsize,
558    /// 当前计划唤醒时刻(相对 graph epoch 的微秒 + 1)；0=尚未排入延迟队列。
559    pub(super) source_wake_deadline_us: AtomicI64,
560    /// 内核主动调用 source_yield 的累计次数。
561    pub(super) source_yield_count: AtomicU64,
562    /// Source 唤醒代次。reset/取消后旧延迟任务会因代次不匹配而失效。
563    pub(super) source_wake_generation: AtomicU64,
564    /// 每个输入口的**时间戳边界**:保证「不会再有时间戳 < bound 的包到来」。
565    /// 这是多输入口对齐的依据 —— 只有确知某口不会再来更早的包,
566    /// 才能安全地在当前最小时间戳上组一次 Process。
567    pub(super) input_bounds: Vec<Mutex<Timestamp>>,
568}
569
570// 安全性:Node 内每个 UnsafeCell<Context> 槽只在被「认领」(从 free_slots 取出而未归还)
571// 期间被访问,认领与归还都在 sched 锁下进行,故同一槽任一时刻只有一个访问者。
572unsafe impl Sync for Node {}
573
574impl Node {
575    /// 取某个 context 槽的可变引用。
576    ///
577    /// # Safety
578    /// 调用者必须**独占持有该槽**(通过在锁下从 `free_slots` 取出而尚未归还),
579    /// 或处于尚未开始调度的阶段(build/start/close,此时 in_flight==0)。
580    #[allow(clippy::mut_from_ref)]
581    pub(super) unsafe fn ctx_slot(&self, slot: usize) -> &mut Context {
582        &mut *self.ctxs[slot].get()
583    }
584
585    pub(super) fn queue_len(&self, port: usize) -> usize {
586        self.input_queues[port]
587            .lock()
588            .expect("queue lock poisoned")
589            .len()
590    }
591    pub(super) fn bound(&self, port: usize) -> Timestamp {
592        *self.input_bounds[port].lock().expect("bound lock poisoned")
593    }
594
595    /// 把某口的时间戳边界向前推进(只增不减)。
596    pub(super) fn advance_bound(&self, port: usize, to: Timestamp) {
597        let mut b = self.input_bounds[port].lock().expect("bound lock poisoned");
598        if to > *b {
599            *b = to;
600        }
601    }
602
603    pub(super) fn front_ts(&self, port: usize) -> Option<Timestamp> {
604        self.input_queues[port]
605            .lock()
606            .expect("queue lock poisoned")
607            .front()
608            .map(|p| p.timestamp())
609    }
610
611    /// 源节点:没有输入口,由内核自行产出(见 docs/design.md §7.4)。
612    pub(super) fn is_source(&self) -> bool {
613        self.input_queues.is_empty()
614    }
615
616    /// 正向(非 back-edge)输入口下标。back-edge 是反馈寄存器,不参与就绪 / 终止 / 对齐判定 ——
617    /// **核心不变式:back-edge 口永不触发 readiness**,故反馈包不会自激无限重跑。
618    pub(super) fn forward_ports(&self) -> impl Iterator<Item = usize> + '_ {
619        (0..self.input_queues.len()).filter(move |&i| !self.input_is_back_edge[i])
620    }
621
622    /// 就绪判定。返回本次的**触发计划**(处理时间戳 + 参与的输入口);`None` 表示还不能跑。
623    ///
624    /// `Sync` 采用时间戳对齐:
625    ///   * `min_packet` = 各非空口队首时间戳的最小值;
626    ///   * `min_bound`  = 各空口边界的最小值(空口若还会来包,最早也不早于其边界);
627    ///   * 仅当 `min_bound > min_packet`,才确知「没有任何口还会送来更早的包」,
628    ///     于是可以在 `min_packet` 上安全地组一次 Process。
629    ///
630    /// 这条判据是多输入口正确性的核心:没有它,Zip 之类算子会把不同时刻的数据配到一起。
631    pub(super) fn readiness(&self) -> Option<Ready> {
632        let n = self.input_queues.len();
633        if n == 0 {
634            // 源节点:无输入口。未自报完成即「可产出」;ts 占位(try_claim 用 seq 覆盖成单调时间戳)。
635            return (!self.source_done.load(Ordering::SeqCst)
636                && !self.source_waiting.load(Ordering::SeqCst))
637            .then_some(Ready {
638                ts: Timestamp::unset(),
639                ports: None,
640                batch: None,
641            });
642        }
643        match &self.policy {
644            InputPolicy::Immediate => {
645                // 不做对齐:任一**正向**口有数据就跑(back-edge 口不触发)
646                let mut min = Timestamp::done();
647                let mut any = false;
648                for i in self.forward_ports() {
649                    if let Some(ts) = self.front_ts(i) {
650                        any = true;
651                        min = min.min(ts);
652                    }
653                }
654                any.then_some(Ready {
655                    ts: min,
656                    ports: None,
657                    batch: None,
658                })
659            }
660            InputPolicy::Sync | InputPolicy::FixedSize { .. } => {
661                self.sync_align(self.forward_ports()).map(|ts| Ready {
662                    ts,
663                    ports: None,
664                    batch: None,
665                })
666            }
667            InputPolicy::SyncSet { sets } => {
668                // 每组独立对齐;取**最早就绪**的那组,本次只带该组的口。
669                let mut best: Option<Ready> = None;
670                for set in sets {
671                    if let Some(ts) = self.sync_align(set.iter().copied()) {
672                        if best.as_ref().is_none_or(|b| ts < b.ts) {
673                            best = Some(Ready {
674                                ts,
675                                ports: Some(set.clone()),
676                                batch: None,
677                            });
678                        }
679                    }
680                }
681                best
682            }
683            InputPolicy::Batch { size } => self.batch_readiness(*size),
684        }
685    }
686
687    /// `batch` 策略的就绪判定:把 `sync` 的对齐**连续跑 `size` 轮**(只偷看、不弹出),
688    /// 算出每个正向口本次该取的包数。
689    ///
690    /// **为什么不是「各口各自攒够 `size` 个」**:那会把 0 号口的第 k 个与 1 号口的第 k 个
691    /// 配成一对,而它们未必是同一帧 —— 图像批与掩码批就此错位,且**不会报任何错**。
692    /// 本项目不接受静默的错误行为,故一批 = `size` 个**对齐元组**,对齐规则与 `sync` 完全同源。
693    ///
694    /// 不足一批时**只有所有正向口都已关闭**才把余量刷出(不可能再来数据了);否则继续等 ——
695    /// 提前交付就是过早切批,那也是一种静默的语义偏差。
696    ///
697    /// 锁:每口**只拿一次**队列锁,把前 `size` 个时间戳快照出来,之后纯内存模拟
698    /// (ADR #36)。前缀在此期间稳定 —— 只有 `try_claim` 会 pop 且全程持 `sched`
699    /// (ADR #30),别的线程只往队尾 push。每口游标每轮至多前进 1、共 `size` 轮,
700    /// 故快照 `size` 个足够。
701    pub(super) fn batch_readiness(&self, size: usize) -> Option<Ready> {
702        let ports: Vec<usize> = self.forward_ports().collect();
703        if ports.is_empty() || size == 0 {
704            return None;
705        }
706
707        let ts_prefix: Vec<Vec<Timestamp>> = ports
708            .iter()
709            .map(|&p| {
710                let q = self.input_queues[p].lock().expect("queue lock poisoned");
711                q.iter().take(size).map(|pk| pk.timestamp()).collect()
712            })
713            .collect();
714
715        let mut take = vec![0usize; ports.len()];
716        let mut first_ts: Option<Timestamp> = None;
717        let mut last_ts = Timestamp::unset();
718        let mut rounds = 0usize;
719
720        while rounds < size {
721            // 一轮对齐,逻辑同 `sync_align`:游标处有包的口贡献 min_packet,
722            // 没包的口贡献它的 bound —— bound 不够大就说明还可能来更早的包,不能定这一轮。
723            let mut min_packet = Timestamp::done();
724            let mut min_bound = Timestamp::done();
725            for (pi, &p) in ports.iter().enumerate() {
726                match ts_prefix[pi].get(take[pi]) {
727                    Some(&ts) => min_packet = min_packet.min(ts),
728                    None => min_bound = min_bound.min(self.bound(p)),
729                }
730            }
731            if min_packet == Timestamp::done() || min_bound <= min_packet {
732                break;
733            }
734            for (pi, _) in ports.iter().enumerate() {
735                if ts_prefix[pi].get(take[pi]) == Some(&min_packet) {
736                    take[pi] += 1;
737                }
738            }
739            first_ts.get_or_insert(min_packet);
740            last_ts = min_packet;
741            rounds += 1;
742        }
743
744        if rounds == 0 {
745            return None;
746        }
747        if rounds < size
748            && !ports
749                .iter()
750                .all(|&p| self.input_closed[p].load(Ordering::SeqCst))
751        {
752            return None; // 还会有数据 → 继续攒,别过早切批
753        }
754
755        Some(Ready {
756            ts: first_ts.unwrap_or_else(Timestamp::unset),
757            ports: None,
758            batch: Some(BatchPlan {
759                take: ports.into_iter().zip(take).collect(),
760                last_ts,
761            }),
762        })
763    }
764
765    /// 在给定的一组输入口上做 sync 对齐,返回对齐到的时间戳(逻辑同 §7.2 的 min_bound>min_packet)。
766    pub(super) fn sync_align(&self, ports: impl Iterator<Item = usize>) -> Option<Timestamp> {
767        let mut min_packet = Timestamp::done();
768        let mut min_bound = Timestamp::done();
769        for i in ports {
770            match self.front_ts(i) {
771                Some(ts) => min_packet = min_packet.min(ts),
772                None => min_bound = min_bound.min(self.bound(i)),
773            }
774        }
775        if min_packet == Timestamp::done() {
776            return None; // 没有任何数据
777        }
778        if min_bound > min_packet {
779            Some(min_packet)
780        } else {
781            None // 某空口还可能送来 <= min_packet 的包,必须再等
782        }
783    }
784
785    pub(super) fn all_inputs_closed_and_drained(&self) -> bool {
786        if self.is_source() {
787            // 源节点没有输入口;只有内核自报完成才算「排空」(否则 (0..0).all() 空真会开图即关)。
788            return self.source_done.load(Ordering::SeqCst);
789        }
790        // 只看正向口:back-edge 口是反馈寄存器,不参与终止判定 —— 否则 A→B→A 里两节点
791        // 互等对方关闭,谁也关不了(终止死锁)。节点靠正向输入排空即可关闭,级联绕环拆解。
792        self.forward_ports()
793            .all(|i| self.input_closed[i].load(Ordering::SeqCst) && self.queue_len(i) == 0)
794    }
795}
lmflow/graph/node.rs

lmflow/graph/
node.rs
