lmflow/executor/
delegating.rs

1use std::collections::VecDeque;
2use std::sync::Mutex;
3use std::time::{Duration, Instant};
4
5use crate::config::StatsLevel;
6use crate::graph::NodeId;
7
8use super::stats::{ExecutorStats, ExecutorStatsSnapshot};
9
10struct QueuedTask {
11    enqueued_at: Option<Instant>,
12    node: NodeId,
13}
14
15/// 委托执行器:一个线程都不拥有,把就绪节点**交还宿主线程**跑。
16///
17/// `submit` 只入队，真正执行发生在宿主进入阻塞接口或主动调用 `Graph::pump_step` 时。
18/// 同一张图的委托任务由图级原子闸门保证零并发。
19pub struct DelegatingExecutor {
20    name: String,
21    queue: Mutex<VecDeque<QueuedTask>>,
22    stats: ExecutorStats,
23}
24
25impl DelegatingExecutor {
26    pub fn new(name: &str) -> Self {
27        Self::new_with_stats(name, StatsLevel::Full)
28    }
29
30    pub(crate) fn new_with_stats(name: &str, stats_level: StatsLevel) -> Self {
31        Self {
32            name: name.to_string(),
33            queue: Mutex::new(VecDeque::new()),
34            stats: ExecutorStats::new(stats_level),
35        }
36    }
37
38    pub fn name(&self) -> &str {
39        &self.name
40    }
41
42    /// 入队待宿主抽取。委托执行器没有已关停状态，故恒返回 `true`。
43    pub fn submit(&self, node: NodeId) -> bool {
44        let mut queue = self.queue.lock().unwrap_or_else(|error| error.into_inner());
45        queue.push_back(QueuedTask {
46            enqueued_at: self.stats.full().then(Instant::now),
47            node,
48        });
49        self.stats.enqueued();
50        true
51    }
52
53    pub fn pending(&self) -> usize {
54        self.queue
55            .lock()
56            .unwrap_or_else(|error| error.into_inner())
57            .len()
58    }
59
60    /// 弹一个待办交给宿主线程跑;`None` = 队列空。
61    pub fn take(&self) -> Option<NodeId> {
62        let mut queue = self.queue.lock().unwrap_or_else(|error| error.into_inner());
63        let task = queue.pop_front()?;
64        self.stats
65            .started(task.enqueued_at.map(|enqueued_at| enqueued_at.elapsed()));
66        Some(task.node)
67    }
68
69    pub fn complete(&self, execution: Option<Duration>) {
70        self.stats.completed(execution);
71    }
72
73    pub(crate) fn stats(&self) -> ExecutorStatsSnapshot {
74        self.stats.snapshot()
75    }
76
77    pub(crate) fn has_pending_work(&self) -> bool {
78        !self
79            .queue
80            .lock()
81            .unwrap_or_else(|error| error.into_inner())
82            .is_empty()
83            || self.stats.has_running()
84    }
85
86    pub(crate) fn queued_nodes(&self) -> Vec<NodeId> {
87        self.queue
88            .lock()
89            .unwrap_or_else(|error| error.into_inner())
90            .iter()
91            .map(|task| task.node)
92            .collect()
93    }
94
95    /// 清空队列(`reset` 用:上一轮的残留不能带进下一轮)。
96    pub fn clear(&self) {
97        let mut queue = self.queue.lock().unwrap_or_else(|error| error.into_inner());
98        let count = queue.len();
99        queue.clear();
100        self.stats.dropped(count);
101    }
102
103    pub fn reset_stats(&self) {
104        self.stats.reset();
105    }
106}
107
108#[cfg(test)]
109mod tests {
110    use super::*;
111
112    #[test]
113    fn queues_fifo_and_never_rejects() {
114        let executor = DelegatingExecutor::new("host");
115        assert!(executor.submit(7));
116        assert!(executor.submit(8));
117        assert_eq!(executor.pending(), 2);
118        assert_eq!(executor.take(), Some(7));
119        assert_eq!(executor.take(), Some(8));
120        assert_eq!(executor.take(), None);
121    }
122
123    #[test]
124    fn clear_drops_pending() {
125        let executor = DelegatingExecutor::new("host");
126        executor.submit(1);
127        executor.submit(2);
128        executor.clear();
129        assert_eq!(executor.pending(), 0);
130    }
131}
lmflow/executor/delegating.rs

lmflow/executor/
delegating.rs
