ek_core/
main_engine.rs

1//! `MainEngine` —— 纯插件宿主:单表注册、(typed)获取、(deps)拓扑 start、
2//! 逆序五步 close、单插件 remove(允许+警告)、运行统计查询。零业务方法。
3//!
4//! close 五步(每插件,逆依赖序依次执行):
5//! ①卸 forwarder(unregister 全部 HandlerId)
6//! ②on_stop 钩子(业务清理)
7//! ③关邮箱(宿主侧 Sender 置 None;全部 send 端消费尽 → 信道断开)
8//! ④worker 排空(crossbeam recv 先取尽缓冲再报 Disconnected)
9//! ⑤join 带超时:超时则告警 + 剥离句柄(线程自然排空收尾,Rust 不杀线程)
10
11use crate::context::PluginContext;
12use crate::error::PluginError;
13use crate::forwarder::{make_forwarder, Forwarder};
14use crate::log_plugin::{LogData, EVENT_LOG, LOG_LEVEL_WARN};
15use crate::plugin::{CLOSE_JOIN_TIMEOUT, Plugin, valid_plugin_name};
16use crate::registry::{EntryStats, PluginEntry, PluginStateCell, PluginStateCode, Registry};
17use crate::worker::spawn_worker;
18use ek_event::{EventEngine, HandlerId};
19use std::any::Any;
20use std::sync::atomic::Ordering;
21use std::sync::{Arc, Mutex, RwLock};
22use std::time::{Duration, Instant};
23
24/// 插件生命周期状态(worker 状态机,AtomicU8 CAS 迁移)。
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum PluginState {
27    Ready,
28    Running,
29    Stopping,
30    Exited,
31    Quarantined,
32}
33
34impl PluginState {
35    pub(crate) fn from_code(c: u8) -> Self {
36        match c {
37            0 => Self::Ready,
38            1 => Self::Running,
39            2 => Self::Stopping,
40            3 => Self::Exited,
41            _ => Self::Quarantined,
42        }
43    }
44    fn from_code_enum(code: PluginStateCode) -> Self {
45        Self::from_code(code as u8)
46    }
47}
48
49/// 插件信息快照(get_plugin)。
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct PluginInfo {
52    pub name: String,
53    pub state: PluginState,
54    pub deps: Vec<String>,
55}
56
57/// 邮箱运行统计(mail_stats)。
58#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
59pub struct MailStats {
60    /// forwarder 溢出丢弃累计。
61    pub spilled: u64,
62    /// handler panic 累计。
63    pub panics: u64,
64}
65
66fn log_event(source: &str, msg: String) -> ek_event::Event {
67    ek_event::Event::with_data(
68        EVENT_LOG,
69        Arc::new(LogData {
70            level: LOG_LEVEL_WARN,
71            source: source.to_string(),
72            msg,
73        }),
74    )
75}
76
77pub struct MainEngine {
78    event: Arc<EventEngine>,
79    registry: Arc<Registry>,
80    /// start() 装配顺序纪录(close 反向使用);None = 未 start。
81    started_order: Mutex<Option<Vec<String>>>,
82    closed: RwLock<bool>,
83}
84
85impl MainEngine {
86    /// 注入式创建:`Some(bus)` 复用外部引擎;`None` 自建(start() 时拉起)。
87    pub fn new(event: Option<Arc<EventEngine>>) -> Self {
88        Self {
89            event: event.unwrap_or_else(|| Arc::new(EventEngine::new())),
90            registry: Arc::new(Registry::new()),
91            started_order: Mutex::new(None),
92            closed: RwLock::new(false),
93        }
94    }
95
96    /// 宿主级引擎句柄(测试/装配工具可用;插件不可见——ctx 不暴露)。
97    pub fn event_engine(&self) -> &Arc<EventEngine> {
98        &self.event
99    }
100
101    fn closed_check(&self) -> Result<(), PluginError> {
102        if *self.closed.read().unwrap() {
103            Err(PluginError::Stopped)
104        } else {
105            Ok(())
106        }
107    }
108
109    fn host_log(&self, msg: String) {
110        let _ = self.event.put(log_event("ek-core", msg));
111    }
112
113    /// 统一格式注册(名/D2、重名、同名自依赖、deps 存在性校验)。
114    pub fn add_plugin(&self, plugin: Arc<impl Plugin + 'static>) -> Result<(), PluginError> {
115        self.closed_check()?;
116        if self.started_order.lock().unwrap().is_some() {
117            return Err(PluginError::HostStarted);
118        }
119        let name = plugin.name().to_string();
120        if !valid_plugin_name(&name) {
121            return Err(PluginError::IllegalName(name));
122        }
123        let deps: Vec<String> = plugin.deps().iter().map(|d| d.to_string()).collect();
124        if deps.iter().any(|d| d == &name) {
125            return Err(PluginError::DepCycle(vec![name]));
126        }
127        let mailbox_cap = plugin.mailbox_cap();
128        let any: Arc<dyn Any + Send + Sync> = Arc::clone(&plugin) as Arc<dyn Any + Send + Sync>;
129        let dyn_plugin: Arc<dyn Plugin> = plugin;
130
131        let mut w = self.registry.state_of.write().unwrap();
132        if w.entries.contains_key(&name) {
133            return Err(PluginError::Duplicate(name));
134        }
135        for d in &deps {
136            if !w.entries.contains_key(d) {
137                return Err(PluginError::UnknownDep(d.clone()));
138            }
139        }
140        let entry = Arc::new(PluginEntry {
141            name: name.clone(),
142            plugin: dyn_plugin,
143            any,
144            state: PluginStateCell::new(PluginStateCode::Ready),
145            mailbox_cap,
146            deps,
147            mailbox: Mutex::new(None),
148            forwarder_ids: Mutex::new(Vec::new()),
149            worker: Mutex::new(None),
150            stats: Arc::new(EntryStats::default()),
151        });
152        w.order.push(entry.name.clone());
153        w.entries.insert(entry.name.clone(), entry);
154        Ok(())
155    }
156
157    /// 状态查询。
158    pub fn get_state(&self, name: &str) -> Option<PluginState> {
159        self.registry.get(name).map(|e| e.state.get())
160    }
161
162    /// 信息快照。
163    pub fn get_plugin(&self, name: &str) -> Option<PluginInfo> {
164        self.registry.get(name).map(|e| PluginInfo {
165            name: e.name.clone(),
166            state: e.state.get(),
167            deps: e.deps.clone(),
168        })
169    }
170
171    /// 类型安全获取(get_typed):同一分配的 Arc<dyn Any> downcast。
172    pub fn get_typed<T: Plugin + 'static>(&self, name: &str) -> Option<Arc<T>> {
173        let entry = self.registry.get(name)?;
174        Arc::downcast::<T>(Arc::clone(&entry.any)).ok()
175    }
176
177    /// 注册名列表(注册序)。
178    pub fn plugin_names(&self) -> Vec<String> {
179        self.registry.names()
180    }
181
182    /// 运行统计(溢出丢弃/panic 计数)。未注册 → None。
183    pub fn mail_stats(&self, name: &str) -> Option<MailStats> {
184        self.registry.get(name).map(|e| MailStats {
185            spilled: e.stats.spilled.load(Ordering::Acquire),
186            panics: e.stats.panics.load(Ordering::Acquire),
187        })
188    }
189
190    /// 启动全部插件:总线 start → deps 拓扑序逐个装配
191    /// (on_init → 建邮箱 → spawn worker → state=Running → 注册 forwarder → on_start)。
192    /// 依赖环 → DepCycle;重复 start 幂等 Ok;已 close → Stopped。
193    pub fn start(&self) -> Result<(), PluginError> {
194        let mut started = self.started_order.lock().unwrap();
195        if started.is_some() {
196            return Ok(());
197        }
198        self.closed_check()?;
199        self.event.start();
200        let order = self.registry.topo_order()?;
201        let mut launched = Vec::with_capacity(order.len());
202        for entry in order {
203            let ctx = Arc::new(PluginContext::new(
204                Arc::clone(&self.event),
205                Arc::clone(&self.registry),
206                entry.name.clone(),
207            ));
208            entry.plugin.on_init(&ctx);
209            let (tx, rx) = crossbeam_channel::bounded::<ek_event::Event>(entry.mailbox_cap);
210            *entry.mailbox.lock().unwrap() = Some(tx.clone());
211            let handler = spawn_worker(
212                Arc::clone(&entry.plugin),
213                rx,
214                entry.state.clone(),
215                Arc::clone(&entry.stats),
216                Arc::clone(&ctx),
217                Arc::clone(&self.event),
218            );
219            *entry.worker.lock().unwrap() = Some(handler);
220            // 先 Running 再装 forwarder:避免 ask 预检在装配窗口把目标误判 Ready
221            let _ = entry.state.cas(PluginStateCode::Ready, PluginStateCode::Running);
222            let mut ids: Vec<HandlerId> = Vec::new();
223            for ty in entry.plugin.routes() {
224                let fwd = Forwarder::new(
225                    tx.clone(),
226                    Arc::clone(&entry.stats),
227                    Arc::clone(&self.event),
228                    entry.name.clone(),
229                );
230                let id = self.event.register(ty.into_owned(), make_forwarder(fwd));
231                ids.push(id);
232            }
233            *entry.forwarder_ids.lock().unwrap() = ids;
234            entry.plugin.on_start();
235            launched.push(entry.name.clone());
236        }
237        *started = Some(launched);
238        Ok(())
239    }
240
241    /// D4 五步闭停单个插件(不摘表)。
242    ///
243    /// on_stop 语义(执行期复审 F1 修正):凡曾装配过的插件(Running/Stopping/Exited/
244    /// Quarantined)close 时都调 on_stop —— 熔断或自然退出的插件也需要业务清理;
245    /// 仅 Ready(从未 start)不调。
246    fn close_one(&self, entry: &Arc<PluginEntry>, deadline: Instant) {
247        let pre_state = entry.state.get();
248        let call_stop = pre_state != PluginState::Ready;
249        // 挂起迁移:Running → Stopping(容错其他状态)
250        let _ = entry.state.cas(PluginStateCode::Running, PluginStateCode::Stopping);
251        // ① 卸 forwarder
252        let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
253        for id in ids {
254            self.event.unregister(id);
255        }
256        // ② on_stop
257        if call_stop {
258            entry.plugin.on_stop();
259        }
260        // ③ 关邮箱(宿主侧 Sender drop → 全部 send 端消费尽)
261        *entry.mailbox.lock().unwrap() = None;
262        // ④⑤ 排空等待 + join(带超时,超时告警 + 剥离)
263        while !entry.stats.done.load(Ordering::Acquire) && Instant::now() < deadline {
264            std::thread::sleep(Duration::from_millis(2));
265        }
266        if let Some(handle) = entry.worker.lock().unwrap().take() {
267            if entry.stats.done.load(Ordering::Acquire) {
268                let _ = handle.join();
269            } else {
270                self.host_log(format!(
271                    "close timeout: worker of {:?} still busy; detach and keep draining",
272                    entry.name
273                ));
274            }
275        }
276        // 兜底:Stopping → Exited(worker 已退出但未来得及 CAS 的场合)
277        let _ = entry.state.cas(PluginStateCode::Stopping, PluginStateCode::Exited);
278    }
279
280    /// 全局闭停:逆 start 序逐插件五步;然后清插件表。总线留给调用方(close_all 停总线)。
281    pub fn close(&self, timeout: Duration) {
282        let order = self.started_order.lock().unwrap().take();
283        *self.closed.write().unwrap() = true;
284        let deadline = Instant::now() + timeout.max(Duration::from_millis(1));
285        if let Some(order) = order {
286            for name in order.iter().rev() {
287                if let Some(entry) = self.registry.get(name) {
288                    self.close_one(&entry, deadline);
289                }
290            }
291        }
292        self.registry.clear();
293    }
294
295    /// close + 停总线(终态)。
296    pub fn close_all(&self, timeout: Duration) {
297        self.close(timeout);
298        self.event.stop();
299    }
300
301    /// 移除单个插件:Running 者执行五步(UNKNOWN 表项报错);仍被其他插件 deps 引用
302    /// 时,eLog 警告(定稿 policy:允许移除 + 警告;依赖方后续 ask 走预检/断连失败)。
303    pub fn remove_plugin(&self, name: &str) -> Result<(), PluginError> {
304        self.closed_check()?;
305        let entry = self
306            .registry
307            .get(name)
308            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
309        let dependents = self.registry.any_depends_on(name);
310        if !dependents.is_empty() {
311            self.host_log(format!(
312                "remove_plugin {:?} still depended on by {:?}",
313                name, dependents
314            ));
315        }
316        let deadline = Instant::now() + CLOSE_JOIN_TIMEOUT;
317        if entry.state.get() == PluginState::from_code_enum(PluginStateCode::Running) {
318            self.close_one(&entry, deadline);
319        } else {
320            // Ready / Exited / Quarantined:摘 forwarder(可能残留于熔断前)+ 关邮箱
321            let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
322            for id in ids {
323                self.event.unregister(id);
324            }
325            *entry.mailbox.lock().unwrap() = None;
326        }
327        self.registry
328            .remove(name)
329            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
330        Ok(())
331    }
332}
ek_core/main_engine.rs

ek_core/
main_engine.rs
