ek_core/
worker.rs

1//! worker(插件串行 actor 线程):邮箱消费 + catch_unwind 隔离 + 熔断(D3)。
2//!
3//! 事件所有权在 worker 循环体内:handler panic unwind 被 catch_unwind 截住后,
4//! 事件随本轮循环作用域自然 drop —— 寻址信封的 reply 随之 drop、信道关闭,
5//! 发起方立即得到 TargetGone(信令依赖于此,勿改为引用投递)。
6
7use crate::lifecycle::EVENT_PLUG_DOWN_PREFIX;
8use crate::plugin::{PANIC_LIMIT, Plugin};
9use crate::registry::{EntryStats, PluginStateCell, PluginStateCode};
10use crossbeam_channel::Receiver;
11use ek_event::{Event, EventEngine};
12use std::panic::{AssertUnwindSafe, catch_unwind};
13use std::sync::Arc;
14use std::sync::atomic::Ordering;
15use std::thread::JoinHandle;
16
17pub(crate) fn spawn_worker(
18    plugin: Arc<dyn Plugin>,
19    rx: Receiver<Event>,
20    state: PluginStateCell,
21    stats: Arc<EntryStats>,
22    bus: Arc<EventEngine>,
23) -> JoinHandle<()> {
24    let name = plugin.name().to_string();
25    std::thread::Builder::new()
26        .name(format!("ek-worker.{name}"))
27        .spawn(move || worker_loop(plugin, rx, state, stats, bus))
28        .expect("ek-core: spawn worker thread")
29}
30
31fn worker_loop(
32    plugin: Arc<dyn Plugin>,
33    rx: Receiver<Event>,
34    state: PluginStateCell,
35    stats: Arc<EntryStats>,
36    bus: Arc<EventEngine>,
37) {
38    // D3 = 连续 panic×3 熔断(复审 F3 修正):处理成功即复位;
39    // stats.panics 为累计统计面,不参与熔断判定。
40    let mut consecutive: u64 = 0;
41    loop {
42        match rx.recv() {
43            Ok(event) => {
44                // 深度计数:工人取走 = 从"积压"变成"在处理",pending −1(0.5.0 口径)。
45                // 用 saturating:这一侧永不因为时序竞争下绕成天文数字(占位在 forwarder 侧)。
46                let _ = stats
47                    .pending
48                    .fetch_update(Ordering::AcqRel, Ordering::Acquire, |v| {
49                        Some(v.saturating_sub(1))
50                    });
51                let result = catch_unwind(AssertUnwindSafe(|| plugin.on_mail(&event)));
52                match result {
53                    Ok(()) => consecutive = 0,
54                    Err(_panic) => {
55                        consecutive += 1;
56                        let _ = stats.panics.fetch_add(1, Ordering::AcqRel);
57                        if consecutive >= PANIC_LIMIT as u64 {
58                            if state.cas(PluginStateCode::Running, PluginStateCode::Quarantined) {
59                                // 运维诊断走 stderr(不进总线,不占类型);结构化通知走 ePlugQDown.<名>
60                                eprintln!(
61                                    "ek-core: plugin {:?} quarantined after {} consecutive handler panics",
62                                    plugin.name(),
63                                    consecutive
64                                );
65                                // 熔断广播(可被监控插件订阅;Q=quarantine)
66                                let _ = bus.put(Event::new(format!(
67                                    "{EVENT_PLUG_DOWN_PREFIX}{}",
68                                    plugin.name()
69                                )));
70                            }
71                            // 检疫即退出:邮箱内残留邮件随之丢弃(drop rx → forwarder 断连短路)
72                            // 深度计数清零:残留帧随 rx 一起没了,不能留下一个永远减不掉的"积压"
73                            stats.pending.store(0, Ordering::Release);
74                            break;
75                        }
76                        // 单次 panic:诊断 stderr;不熔断、不广播
77                        eprintln!(
78                            "ek-core: plugin {:?} handler panicked ({consecutive}/{PANIC_LIMIT})",
79                            plugin.name()
80                        );
81                    }
82                }
83                // event 随本轮作用域 drop → 寻址信封 reply drop → 请求方即时感知
84                drop(event);
85            }
86            // 排空 + 断连:close 已把全部 send 端消费尽
87            Err(_disconnected) => {
88                state.finish();
89                break;
90            }
91        }
92    }
93    stats.done.store(true, Ordering::Release);
94}
ek_core/worker.rs

ek_core/
worker.rs
