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

ek_core/
main_engine.rs
