ek_core/
registry.rs

1//! 注册表(内部):单表统一 add / get / typed get / remove;
2//! 状态原子 CAS;deps 快照;start 拓扑序。
3
4use crate::error::PluginError;
5use crate::main_engine::PluginState;
6use ek_event::{Event, HandlerId};
7use std::any::Any;
8use std::collections::{HashMap, VecDeque};
9use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
10use std::sync::{Arc, Mutex, RwLock};
11use std::thread::JoinHandle;
12
13use crossbeam_channel::Sender;
14
15/// 每插件运行统计(供宿主查询/测试断言/告警)。
16#[derive(Default)]
17pub(crate) struct EntryStats {
18    /// forwarder 溢出丢弃累计(邮箱满)。
19    pub spilled: AtomicU64,
20    /// handler panic 累计。
21    pub panics: AtomicU64,
22    /// worker 退出标志(join-with-deadline 用,D4)。
23    pub done: AtomicBool,
24}
25
26/// 注册表的一条记录(内部,不公开)。
27pub(crate) struct PluginEntry {
28    pub name: String,
29    /// 供 worker/on_mail 与生命周期钩子调用的 trait 对象。
30    pub plugin: Arc<dyn crate::plugin::Plugin>,
31    /// 指向同一分配的 `Arc<dyn Any + Send + Sync>`:get_typed 的 downcast 底座。
32    pub any: Arc<dyn Any + Send + Sync>,
33    pub state: PluginStateCell,
34    pub mailbox_cap: usize,
35    pub deps: Vec<String>,
36    /// 宿主侧邮箱发端(持有一份;close 第③步置 None → 全部 Sender 消费尽 → 断连)。
37    pub mailbox: Mutex<Option<Sender<Event>>>,
38    /// start 装配的 forwarder 注册 id(close/remove 第①步逐个卸)。
39    pub forwarder_ids: Mutex<Vec<HandlerId>>,
40    /// worker 句柄(close 第⑤步 join,超时则弃-剥离)。
41    pub worker: Mutex<Option<JoinHandle<()>>>,
42    pub stats: Arc<EntryStats>,
43}
44
45/// 插件状态(AtomicU8 CAS 迁移,非法迁移拒绝)。
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47pub enum PluginStateCode {
48    Ready = 0,
49    Running = 1,
50    Stopping = 2,
51    Exited = 3,
52    Quarantined = 4,
53}
54
55/// 状态原子单元(锁-free;Clone 共享同一原子底座)。
56#[derive(Clone)]
57pub(crate) struct PluginStateCell(Arc<AtomicU8>);
58
59impl PluginStateCell {
60    pub fn new(s: PluginStateCode) -> Self {
61        Self(Arc::new(AtomicU8::new(s as u8)))
62    }
63    pub fn get(&self) -> PluginState {
64        PluginState::from_code(self.0.load(Ordering::Acquire))
65    }
66    pub(crate) fn cas(&self, from: PluginStateCode, to: PluginStateCode) -> bool {
67        self.0
68            .compare_exchange(from as u8, to as u8, Ordering::AcqRel, Ordering::Acquire)
69            .is_ok()
70    }
71    /// worker 自然结束:Running/Stopping → Exited。
72    pub(crate) fn finish(&self) -> bool {
73        self.cas(PluginStateCode::Running, PluginStateCode::Exited)
74            || self.cas(PluginStateCode::Stopping, PluginStateCode::Exited)
75    }
76}
77
78/// 内部注册表集中区。
79pub(crate) struct RegistryInner {
80    pub entries: HashMap<String, Arc<PluginEntry>>,
81    /// 注册顺序(确定性、拓扑并列时的排序依据)。
82    pub order: Vec<String>,
83}
84
85pub(crate) struct Registry {
86    pub state_of: RwLock<RegistryInner>,
87}
88
89impl Registry {
90    pub fn new() -> Self {
91        Self { state_of: RwLock::new(RegistryInner { entries: HashMap::new(), order: Vec::new() }) }
92    }
93
94    pub fn get(&self, name: &str) -> Option<Arc<PluginEntry>> {
95        self.state_of.read().unwrap().entries.get(name).cloned()
96    }
97
98    pub fn remove(&self, name: &str) -> Option<Arc<PluginEntry>> {
99        let mut w = self.state_of.write().unwrap();
100        w.entries.remove(name).inspect(|_| {
101            if let Some(i) = w.order.iter().position(|n| n == name) {
102                w.order.remove(i);
103            }
104        })
105    }
106
107    pub fn names(&self) -> Vec<String> {
108        self.state_of.read().unwrap().order.clone()
109    }
110
111    pub fn any_depends_on(&self, target: &str) -> Vec<String> {
112        let r = self.state_of.read().unwrap();
113        r.order
114            .iter()
115            .filter_map(|n| r.entries.get(n))
116            .filter(|e| e.deps.iter().any(|d| d == target))
117            .map(|e| e.name.clone())
118            .collect()
119    }
120
121    pub fn clear(&self) {
122        let mut w = self.state_of.write().unwrap();
123        w.entries.clear();
124        w.order.clear();
125    }
126
127    /// start() 的依赖拓扑序(Kahn;并列按注册序;环 → Err)。
128    pub fn topo_order(&self) -> Result<Vec<Arc<PluginEntry>>, PluginError> {
129        let r = self.state_of.read().unwrap();
130        let mut indeg: HashMap<String, usize> = HashMap::new();
131        let mut dependents: HashMap<String, Vec<String>> = HashMap::new();
132        for name in &r.order {
133            indeg.entry(name.clone()).or_insert(0);
134        }
135        for name in &r.order {
136            let e = &r.entries[name];
137            for d in &e.deps {
138                // add 期已校验存在;start 前可能被 remove → 悬空依赖按缺失报错
139                if !r.entries.contains_key(d) {
140                    return Err(PluginError::UnknownDep(d.clone()));
141                }
142                *indeg.entry(name.clone()).or_insert(0) += 1;
143                dependents.entry(d.clone()).or_default().push(name.clone());
144            }
145        }
146        let mut queue: VecDeque<String> = r
147            .order
148            .iter()
149            .filter(|n| indeg[*n] == 0)
150            .cloned()
151            .collect();
152        let mut out = Vec::with_capacity(r.order.len());
153        while let Some(n) = queue.pop_front() {
154            let e = Arc::clone(&r.entries[&n]);
155            out.push(e);
156            if let Some(deps_list) = dependents.get(&n) {
157                for dep_name in deps_list {
158                    let d = indeg.get_mut(dep_name).unwrap();
159                    *d -= 1;
160                    if *d == 0 {
161                        queue.push_back(dep_name.clone());
162                    }
163                }
164            }
165        }
166        if out.len() < r.order.len() {
167            let cycle: Vec<String> = r.order.iter().filter(|n| indeg[*n] > 0).cloned().collect();
168            return Err(PluginError::DepCycle(cycle));
169        }
170        Ok(out)
171    }
172}
ek_core/registry.rs

ek_core/
registry.rs
