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::{Forwarder, make_forwarder};
14use crate::lifecycle::{
15    EVENT_PLUG_DOWN, EVENT_PLUG_DOWN_EVENT_PREFIX, EVENT_PLUG_UP, EVENT_PLUG_UP_PREFIX,
16};
17use crate::plugin::{CLOSE_JOIN_TIMEOUT, Plugin, valid_plugin_name};
18use crate::registry::{EntryStats, PluginEntry, PluginStateCell, PluginStateCode, Registry};
19use crate::service::PlugNotice;
20use crate::worker::spawn_worker;
21use ek_event::{EventEngine, HandlerId};
22use std::any::Any;
23use std::sync::atomic::Ordering;
24use std::sync::{Arc, Mutex, RwLock};
25use std::time::{Duration, Instant};
26
27/// 插件生命周期状态(worker 状态机,AtomicU8 CAS 迁移)。
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum PluginState {
30    Ready,
31    Running,
32    Stopping,
33    Exited,
34    Quarantined,
35}
36
37impl PluginState {
38    pub(crate) fn from_code(c: u8) -> Self {
39        match c {
40            0 => Self::Ready,
41            1 => Self::Running,
42            2 => Self::Stopping,
43            3 => Self::Exited,
44            _ => Self::Quarantined,
45        }
46    }
47    fn from_code_enum(code: PluginStateCode) -> Self {
48        Self::from_code(code as u8)
49    }
50}
51
52/// 插件信息快照(get_plugin / ctx.plugins())。
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct PluginInfo {
55    pub name: String,
56    /// 插件申报版本(空串 = 未声明);A 侧版本策略据此选轨(F5:单一披露面)。
57    pub version: String,
58    pub state: PluginState,
59    pub deps: Vec<String>,
60}
61
62/// 邮箱运行统计(mail_stats)。
63#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
64pub struct MailStats {
65    /// forwarder 溢出丢弃累计。
66    pub spilled: u64,
67    /// handler panic 累计。
68    pub panics: u64,
69}
70
71pub struct MainEngine {
72    event: Arc<EventEngine>,
73    registry: Arc<Registry>,
74    /// start() 装配顺序纪录(close 反向使用);None = 未 start。
75    started_order: Mutex<Option<Vec<String>>>,
76    closed: RwLock<bool>,
77}
78
79impl MainEngine {
80    /// 注入式创建:`Some(bus)` 复用外部引擎;`None` 自建(start() 时拉起)。
81    pub fn new(event: Option<Arc<EventEngine>>) -> Self {
82        Self {
83            event: event.unwrap_or_else(|| Arc::new(EventEngine::new())),
84            registry: Arc::new(Registry::new()),
85            started_order: Mutex::new(None),
86            closed: RwLock::new(false),
87        }
88    }
89
90    /// 宿主级引擎句柄(测试/装配工具可用;插件不可见——ctx 不暴露)。
91    pub fn event_engine(&self) -> &Arc<EventEngine> {
92        &self.event
93    }
94
95    fn closed_check(&self) -> Result<(), PluginError> {
96        if *self.closed.read().unwrap() {
97            Err(PluginError::Stopped)
98        } else {
99            Ok(())
100        }
101    }
102
103    /// 统一格式注册(D2:名/重名/同名自依赖;add 期【不校验 deps 存在性】——
104    /// 新 D1 注册自由,悬空 deps 由 start 期把关)。
105    ///
106    /// **运行期动态 add(0.2.0)**:start 之后仍可随时注入新插件——登记 → 织入
107    /// 能力索引 → 立即走 launch_one 装配上线 → 记入 started_order(close 反序覆盖);
108    /// 并广播 `ePlugUp` / `ePlugUp.<名>`,迟绑定依赖方据此自主发现。
109    pub fn add_plugin(&self, plugin: Arc<impl Plugin + 'static>) -> Result<(), PluginError> {
110        self.closed_check()?;
111        let name = plugin.name().to_string();
112        if !valid_plugin_name(&name) {
113            return Err(PluginError::IllegalName(name));
114        }
115        let deps: Vec<String> = plugin.deps().iter().map(|d| d.to_string()).collect();
116        if deps.iter().any(|d| d == &name) {
117            return Err(PluginError::DepCycle(vec![name]));
118        }
119        let mailbox_cap = plugin.mailbox_cap();
120        let any: Arc<dyn Any + Send + Sync> = Arc::clone(&plugin) as Arc<dyn Any + Send + Sync>;
121        let dyn_plugin: Arc<dyn Plugin> = plugin;
122        // F8 纪律:services() 为纯声明;在拿锁前读一次抽 id
123        let svc_ids: Vec<String> = {
124            let mut ids = Vec::new();
125            for d in dyn_plugin.services() {
126                if !ids.contains(&d.id) {
127                    ids.push(d.id);
128                }
129            }
130            ids
131        };
132
133        let mut w = self.registry.state_of.write().unwrap();
134        if w.entries.contains_key(&name) {
135            return Err(PluginError::Duplicate(name));
136        }
137        let entry = Arc::new(PluginEntry {
138            name: name.clone(),
139            plugin: dyn_plugin,
140            any,
141            state: PluginStateCell::new(PluginStateCode::Ready),
142            mailbox_cap,
143            deps,
144            mailbox: Mutex::new(None),
145            forwarder_ids: Mutex::new(Vec::new()),
146            worker: Mutex::new(None),
147            stats: Arc::new(EntryStats::default()),
148        });
149        w.order.push(entry.name.clone());
150        w.entries.insert(entry.name.clone(), entry);
151        drop(w);
152        self.registry.index_put(&name, &svc_ids);
153        self.registry.invalidate(&name);
154
155        // 动态 add:start 已发生 → 立即装配上线(与 start 共用 launch_one)
156        let mut started = self.started_order.lock().unwrap();
157        if started.is_some() {
158            let entry = self.registry.get(&name).expect("just inserted");
159            self.launch_one(&entry);
160            started.as_mut().expect("checked Some").push(name);
161        }
162        Ok(())
163    }
164
165    /// 状态查询。
166    pub fn get_state(&self, name: &str) -> Option<PluginState> {
167        self.registry.get(name).map(|e| e.state.get())
168    }
169
170    /// 信息快照。
171    pub fn get_plugin(&self, name: &str) -> Option<PluginInfo> {
172        self.registry.info_of(name)
173    }
174
175    /// 类型安全获取(get_typed):同一分配的 Arc<dyn Any> downcast。
176    pub fn get_typed<T: Plugin + 'static>(&self, name: &str) -> Option<Arc<T>> {
177        let entry = self.registry.get(name)?;
178        Arc::downcast::<T>(Arc::clone(&entry.any)).ok()
179    }
180
181    /// 注册名列表(注册序)。
182    pub fn plugin_names(&self) -> Vec<String> {
183        self.registry.names()
184    }
185
186    /// 运行统计(溢出丢弃/panic 计数)。未注册 → None。
187    pub fn mail_stats(&self, name: &str) -> Option<MailStats> {
188        self.registry.get(name).map(|e| MailStats {
189            spilled: e.stats.spilled.load(Ordering::Acquire),
190            panics: e.stats.panics.load(Ordering::Acquire),
191        })
192    }
193
194    /// 启动全部插件:总线 start → deps 拓扑序逐个装配
195    /// (on_init → 建邮箱 → spawn worker → state=Running → 注册 forwarder → on_start
196    /// → 广播 ePlugUp)。
197    /// 依赖环/悬空 deps → DepCycle/UnknownDep;重复 start 幂等 Ok;已 close → Stopped。
198    pub fn start(&self) -> Result<(), PluginError> {
199        let mut started = self.started_order.lock().unwrap();
200        if started.is_some() {
201            return Ok(());
202        }
203        self.closed_check()?;
204        self.event.start();
205        let order = self.registry.topo_order()?;
206        let mut launched = Vec::with_capacity(order.len());
207        for entry in order {
208            self.launch_one(&entry);
209            launched.push(entry.name.clone());
210        }
211        *started = Some(launched);
212        Ok(())
213    }
214
215    /// 单插件装配(start 与动态 add 共用的最小闭包):
216    /// on_init → 建邮箱 → spawn worker → Running → 注册 forwarder → on_start
217    /// → 广播 `ePlugUp.<名>`(后缀版,载荷=版本串)+ `ePlugUp`(固定版,
218    /// 载荷=PlugNotice,供不认识插件名的巡检者订阅)。
219    fn launch_one(&self, entry: &Arc<PluginEntry>) {
220        let ctx = Arc::new(PluginContext::new(
221            Arc::clone(&self.event),
222            Arc::clone(&self.registry),
223            entry.name.clone(),
224        ));
225        entry.plugin.on_init(&ctx);
226        let (tx, rx) = crossbeam_channel::bounded::<ek_event::Event>(entry.mailbox_cap);
227        *entry.mailbox.lock().unwrap() = Some(tx.clone());
228        let handler = spawn_worker(
229            Arc::clone(&entry.plugin),
230            rx,
231            entry.state.clone(),
232            Arc::clone(&entry.stats),
233            Arc::clone(&self.event),
234        );
235        *entry.worker.lock().unwrap() = Some(handler);
236        // 先 Running 再装 forwarder:避免 ask 预检在装配窗口把目标误判 Ready
237        let _ = entry
238            .state
239            .cas(PluginStateCode::Ready, PluginStateCode::Running);
240        let mut ids: Vec<HandlerId> = Vec::new();
241        for ty in entry.plugin.routes() {
242            let fwd = Forwarder::new(tx.clone(), Arc::clone(&entry.stats), entry.name.clone());
243            let id = self.event.register(ty.into_owned(), make_forwarder(fwd));
244            ids.push(id);
245        }
246        *entry.forwarder_ids.lock().unwrap() = ids;
247        entry.plugin.on_start();
248
249        let version = entry.plugin.version().into_owned();
250        let _ = self.event.put(ek_event::Event::with_data(
251            format!("{EVENT_PLUG_UP_PREFIX}{}", entry.name),
252            Arc::new(version.clone()),
253        ));
254        let _ = self.event.put(ek_event::Event::with_data(
255            EVENT_PLUG_UP,
256            Arc::new(PlugNotice {
257                name: entry.name.clone(),
258                version,
259            }),
260        ));
261    }
262
263    /// D4 五步闭停单个插件(不摘表)。
264    ///
265    /// on_stop 语义(执行期复审 F1 修正):凡曾装配过的插件(Running/Stopping/Exited/
266    /// Quarantined)close 时都调 on_stop —— 熔断或自然退出的插件也需要业务清理;
267    /// 仅 Ready(从未 start)不调。
268    fn close_one(&self, entry: &Arc<PluginEntry>, deadline: Instant) {
269        let pre_state = entry.state.get();
270        let call_stop = pre_state != PluginState::Ready;
271        // 挂起迁移:Running → Stopping(容错其他状态)
272        let _ = entry
273            .state
274            .cas(PluginStateCode::Running, PluginStateCode::Stopping);
275        // ① 卸 forwarder
276        let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
277        for id in ids {
278            self.event.unregister(id);
279        }
280        // ② on_stop
281        if call_stop {
282            entry.plugin.on_stop();
283        }
284        // ③ 关邮箱(宿主侧 Sender drop → 全部 send 端消费尽)
285        *entry.mailbox.lock().unwrap() = None;
286        // ④⑤ 排空等待 + join(带超时,超时告警 + 剥离)
287        while !entry.stats.done.load(Ordering::Acquire) && Instant::now() < deadline {
288            std::thread::sleep(Duration::from_millis(2));
289        }
290        if let Some(handle) = entry.worker.lock().unwrap().take() {
291            if entry.stats.done.load(Ordering::Acquire) {
292                let _ = handle.join();
293            } else {
294                // 运维诊断走 stderr(0.3.0:不进总线,不占类型)
295                eprintln!(
296                    "ek-core: close timeout: worker of {:?} still busy; detach and keep draining",
297                    entry.name
298                );
299            }
300        }
301        // 兜底:Stopping → Exited(worker 已退出但未来得及 CAS 的场合)
302        let _ = entry
303            .state
304            .cas(PluginStateCode::Stopping, PluginStateCode::Exited);
305    }
306
307    /// 全局闭停:逆 start 序逐插件五步;然后清插件表。总线留给调用方(close_all 停总线)。
308    pub fn close(&self, timeout: Duration) {
309        let order = self.started_order.lock().unwrap().take();
310        *self.closed.write().unwrap() = true;
311        let deadline = Instant::now() + timeout.max(Duration::from_millis(1));
312        if let Some(order) = order {
313            for name in order.iter().rev() {
314                if let Some(entry) = self.registry.get(name) {
315                    self.close_one(&entry, deadline);
316                }
317            }
318        }
319        self.registry.clear();
320    }
321
322    /// close + 停总线(终态)。
323    pub fn close_all(&self, timeout: Duration) {
324        self.close(timeout);
325        self.event.stop();
326    }
327
328    /// 移除单个插件:Running 者执行五步(UNKNOWN 表项报错);仍被其他插件 deps 引用
329    /// 时 stderr 警告(定稿 policy:允许移除 + 警告;依赖方后续 ask 走预检/断连失败)。
330    ///
331    /// 0.2.0:同步修剪能力索引(F3:防幽灵提供方)+ 作废缓存,并发
332    /// `ePlugDown`(固定版,PlugNotice)+ `ePlugDown.<名>`(后缀版,版本串)。
333    /// 注:close()(宿主整体下架)不发逐插件 ePlugDown(审核 F7:避免拆除期事件风暴)。
334    pub fn remove_plugin(&self, name: &str) -> Result<(), PluginError> {
335        self.closed_check()?;
336        let entry = self
337            .registry
338            .get(name)
339            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
340        let dependents = self.registry.any_depends_on(name);
341        if !dependents.is_empty() {
342            eprintln!(
343                "ek-core: remove_plugin {:?} still depended on by {:?}",
344                name, dependents
345            );
346        }
347        let deadline = Instant::now() + CLOSE_JOIN_TIMEOUT;
348        if entry.state.get() == PluginState::from_code_enum(PluginStateCode::Running) {
349            self.close_one(&entry, deadline);
350        } else {
351            // Ready / Exited / Quarantined:摘 forwarder(可能残留于熔断前)+ 关邮箱
352            let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
353            for id in ids {
354                self.event.unregister(id);
355            }
356            *entry.mailbox.lock().unwrap() = None;
357        }
358        let version = entry.plugin.version().into_owned();
359        self.registry
360            .remove(name)
361            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
362        // 摘表成功才动索引(修剪幽灵;F3)+ 失效缓存 + 移除广播
363        self.registry.index_remove(name);
364        self.registry.invalidate(name);
365        let _ = self.event.put(ek_event::Event::with_data(
366            format!("{EVENT_PLUG_DOWN_EVENT_PREFIX}{name}"),
367            Arc::new(version.clone()),
368        ));
369        let _ = self.event.put(ek_event::Event::with_data(
370            EVENT_PLUG_DOWN,
371            Arc::new(PlugNotice {
372                name: name.to_string(),
373                version,
374            }),
375        ));
376        Ok(())
377    }
378}
ek_core/main_engine.rs

ek_core/
main_engine.rs
