lmflow/
executor.rs

1//! 执行器:把「就绪节点」派给某个执行器执行。
2//!
3//! 两种执行器,由 YAML 的 `executors[].type` 选:
4//!  * [`ThreadPool`] —— 拥有自己的工作线程,真并发。**默认执行器就是它**。
5//!  * [`DelegatingExecutor`] —— 一个线程都不拥有,把任务交还**宿主线程**跑。
6//!
7//! 两者对调度器是同一个东西([`Executor`]),`GraphInner::dispatch_task` 不为谁分叉。
8//! 具体实现分别位于 `executor/thread_pool.rs` 与 `executor/delegating.rs`。
9
10use std::sync::Weak;
11use std::time::Duration;
12
13use crate::graph::{GraphInner, NodeId};
14
15mod delegating;
16mod platform;
17mod stats;
18mod thread_pool;
19
20pub use delegating::DelegatingExecutor;
21pub(crate) use stats::ExecutorStatsSnapshot;
22pub use thread_pool::ThreadPool;
23
24/// 引擎隐式默认执行器的名字。
25///
26/// 节点不写 `executor` 就归到它。它**完全由引擎持有**:恒是一个按 CPU 核数开线程的
27/// 线程池,不绑核、不设实时优先级,YAML 里无从干涉 —— 想控制这些就自己声明一个具名池,
28/// 把节点用 `executor:` 指过去。
29///
30/// 因此这个名字在 `executors` 里是**保留的**,声明即报错:否则一张图里会同时出现
31/// 两个 `default`。`executors` 里写的一律是宿主自己的执行器,且必须有名字。
32pub const DEFAULT_EXECUTOR_NAME: &str = "default";
33
34/// 一个执行器。
35///
36/// 用 `enum` 而不是 `Box<dyn Executor>`:`dispatch_task` 在热路径上
37/// (`schedule_node` 的 `while try_claim` 每轮都调一次),静态分发省掉虚表;
38/// 而且「关停 + join 必须发生在动节点之前」这条约定要看得清具体类型。
39pub enum Executor {
40    /// 拥有自己工作线程的线程池(YAML `type: "ThreadPoolExecutor"`)。
41    Pool(ThreadPool),
42    /// 不拥有线程,交还宿主线程(YAML `type: "DelegatingExecutor"`)。
43    Delegating(DelegatingExecutor),
44}
45
46impl Executor {
47    pub fn name(&self) -> &str {
48        match self {
49            Self::Pool(pool) => pool.name(),
50            Self::Delegating(delegating) => delegating.name(),
51        }
52    }
53
54    /// 是否把任务交还宿主线程(而不是自有工作线程)。
55    pub fn is_delegating(&self) -> bool {
56        matches!(self, Self::Delegating(_))
57    }
58
59    /// **自有**工作线程数。委托执行器返回 0 —— 它一个线程都不拥有。
60    pub fn num_threads(&self) -> usize {
61        match self {
62            Self::Pool(pool) => pool.num_threads(),
63            Self::Delegating(_) => 0,
64        }
65    }
66
67    /// CPU 亲和力核列表(空 = 不绑)。仅用于内省 / 可视化。
68    pub fn affinity(&self) -> &[usize] {
69        match self {
70            Self::Pool(pool) => pool.affinity(),
71            Self::Delegating(_) => &[],
72        }
73    }
74
75    /// 实时优先级(0 = 普通分时)。仅用于内省 / 可视化。
76    pub fn priority(&self) -> i32 {
77        match self {
78            Self::Pool(pool) => pool.priority(),
79            Self::Delegating(_) => 0,
80        }
81    }
82
83    /// 拉起工作线程。委托执行器是 no-op:宿主线程早就在跑,不由引擎创建。
84    pub fn start(&self, graph: Weak<GraphInner>) {
85        if let Self::Pool(pool) = self {
86            pool.start(graph);
87        }
88    }
89
90    /// 投递一个就绪节点。返回 `false` 表示没收下,调用方须善后(撤销已取得的令牌)。
91    pub fn submit(&self, node: NodeId) -> bool {
92        match self {
93            Self::Pool(pool) => pool.submit(node),
94            Self::Delegating(delegating) => delegating.submit(node),
95        }
96    }
97
98    pub fn submit_source_wake(&self, node: NodeId, generation: u64, delay: Duration) -> bool {
99        match self {
100            Self::Pool(pool) => pool.submit_source_wake(node, generation, delay),
101            Self::Delegating(_) => false,
102        }
103    }
104
105    pub fn pending(&self) -> usize {
106        match self {
107            Self::Pool(pool) => pool.pending(),
108            Self::Delegating(delegating) => delegating.pending(),
109        }
110    }
111
112    /// 关停并 join。幂等。委托执行器是 no-op(无线程可 join)。
113    pub fn shutdown(&self) {
114        if let Self::Pool(pool) = self {
115            pool.shutdown();
116        }
117    }
118
119    /// 弹一个**待宿主执行**的任务;线程池恒返回 `None`(它的任务由自己的 worker 取)。
120    pub fn take_delegated(&self) -> Option<NodeId> {
121        match self {
122            Self::Pool(_) => None,
123            Self::Delegating(delegating) => delegating.take(),
124        }
125    }
126
127    pub fn complete_delegated(&self, execution: Option<Duration>) {
128        if let Self::Delegating(delegating) = self {
129            delegating.complete(execution);
130        }
131    }
132
133    pub(crate) fn stats(&self) -> ExecutorStatsSnapshot {
134        match self {
135            Self::Pool(pool) => pool.stats(),
136            Self::Delegating(delegating) => delegating.stats(),
137        }
138    }
139
140    pub(crate) fn has_pending_work(&self) -> bool {
141        match self {
142            Self::Pool(pool) => pool.has_pending_work(),
143            Self::Delegating(delegating) => delegating.has_pending_work(),
144        }
145    }
146
147    pub(crate) fn queued_nodes(&self) -> Vec<NodeId> {
148        match self {
149            Self::Pool(pool) => pool.queued_nodes(),
150            Self::Delegating(delegating) => delegating.queued_nodes(),
151        }
152    }
153
154    /// 清空待宿主执行的队列(`reset` 用);线程池是 no-op。
155    pub fn clear_delegated(&self) {
156        if let Self::Delegating(delegating) = self {
157            delegating.clear();
158        }
159    }
160
161    pub fn reset_run_state(&self) {
162        match self {
163            Self::Pool(pool) => {
164                pool.clear_delayed();
165                pool.reset_stats();
166            }
167            Self::Delegating(delegating) => {
168                delegating.clear();
169                delegating.reset_stats();
170            }
171        }
172    }
173
174    pub fn clear_delayed(&self) {
175        if let Self::Pool(pool) = self {
176            pool.clear_delayed();
177        }
178    }
179}
180
181#[cfg(test)]
182mod tests {
183    use super::*;
184
185    #[test]
186    fn delegating_submit_survives_shutdown() {
187        let executor = Executor::Delegating(DelegatingExecutor::new("host"));
188        executor.shutdown();
189        assert!(
190            executor.submit(1),
191            "委托执行器没有『已关停』状态 —— 宿主线程不由引擎拉起,也无从 join"
192        );
193    }
194
195    #[test]
196    fn executor_kinds_report_their_shape() {
197        let pool = Executor::Pool(ThreadPool::new("cpu", 4, vec![2, 3], 10));
198        let host = Executor::Delegating(DelegatingExecutor::new(DEFAULT_EXECUTOR_NAME));
199
200        assert!(!pool.is_delegating());
201        assert_eq!(pool.name(), "cpu");
202        assert_eq!(pool.num_threads(), 4);
203        assert_eq!(pool.affinity(), &[2, 3]);
204        assert_eq!(pool.priority(), 10);
205        assert_eq!(pool.take_delegated(), None);
206
207        assert!(host.is_delegating());
208        assert_eq!(host.name(), DEFAULT_EXECUTOR_NAME);
209        assert_eq!(host.num_threads(), 0);
210        assert!(host.affinity().is_empty());
211        assert_eq!(host.priority(), 0);
212        host.submit(5);
213        assert_eq!(host.take_delegated(), Some(5));
214    }
215}
lmflow/executor.rs

lmflow/
executor.rs
