mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-29 22:12:52 +00:00
fix(plugin): StopAll 并行停插件 —— 修关停必然超时被 SIGKILL
## 现象 systemd 每次都报 `State 'stop-sigterm' timed out. Killing.` 进程组里 23 个子进程插件**全退完了**,最后那条 `[homed] stopped` 仍打不出来,然后被 SIGKILL。 ## 根因(算出来的,不是猜的) 单个插件的 Stop 最坏预算: 5s(CallContext plugin.stop)+ 5s(等 exited)+ 2s(Kill 后收割)= 12s 串行停 23 个 ⇒ 23 × 12s = 276s,而 systemd 只给 90s。 ⇒ 关停必然超时。线上每一条 stop 记录都是 timed out,无一例外。 ## 改法 StopAll 改为并行:取插件快照 + 各自的 SDK 句柄后**立即释放 registry 锁**, 每个插件一个 goroutine,等全部完成再释放共享段。 三处必须小心的点(都是并行化引入的新风险): 1. **先释放 registry 锁再并行停**。p.Stop() 会触发 markExited → onExit → ReclaimOwner,那条链要读共享内存段。 持着锁并行跑,若某插件的 onExit 需要拿 registry 锁(摘通道等) 就是自死锁。 2. **stop handler 的快照要在清空 sdkRefs 之前取**。handler 挂在 PluginSDK 上(r.sdkRefs),先清空就再也拿不到了。 3. **单个插件 panic 不带崩关停**(那会让剩下的插件全停不掉), 也不静默吞(留日志)。 ## 判据:5 条 + 3 组变异 + race - TestStopAllStopsInParallel:8 个插件各 120ms,串行 960ms / 并行 120ms。 判据直接量耗时,串行实现必然变红(实测串行时 962ms)。 - TestStopAllSurvivesPanickingPlugin:panic 不外冒、其它插件照停、不死锁 (带 10s 超时,死锁会超时而不是挂住测试) - TestStopAllFreezesAutoRestart / StopsEachPluginExactlyOnce: 冻结自动重启、每个插件恰好 Stop 一次(重复会二次释放共享段) - TestStopAllRunsStopHandlerBeforeStop:handler 必须先于 Stop, 走真实的 PluginSDK.RegisterStopHandler 路径 变异:退回串行 → 判红并打出实测耗时;去掉 handler 调用 → 顺序判红; 假装并行只清空 → 4 条判红。 `-race` 通过(并行化必须验锁,这是本次改动的头号风险)。 全量 41 包绿。
This commit is contained in:
190
internal/plugin/stopall_parallel_test.go
Normal file
190
internal/plugin/stopall_parallel_test.go
Normal file
@ -0,0 +1,190 @@
|
||||
package plugin
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
||||
)
|
||||
|
||||
// ===== StopAll 并行化 =====
|
||||
//
|
||||
// 串行停 23 个插件、每个最坏 5s(plugin.stop 调用)+5s(等退出)+2s(收割)
|
||||
// = 最坏 276s,而 systemd 只给 90s ⇒ 关停几乎必然被 SIGKILL。
|
||||
// 线上实测:每次 stop 都 "State 'stop-sigterm' timed out",
|
||||
// 进程组里 23 个插件全退完了,最后那条 [homed] stopped 仍打不出来。
|
||||
//
|
||||
// 并行后最坏约等于单个插件的预算(约 12s),而不是 N 倍。
|
||||
//
|
||||
// ★ 并行化最大的风险是死锁与重复释放:Stop 会触发 markExited →
|
||||
// onExit → ReclaimOwner,后者要读共享内存段。所以判据同时盯
|
||||
// "真的并行"与"不死锁、不错杀"。
|
||||
|
||||
// stubPlugin 是最小可用插件。
|
||||
type stubPlugin struct {
|
||||
name string
|
||||
delay time.Duration
|
||||
stopped atomic.Bool
|
||||
panicOnStop bool
|
||||
stopCount atomic.Int32
|
||||
}
|
||||
|
||||
func newStubPlugin(name string, delay time.Duration) *stubPlugin {
|
||||
return &stubPlugin{name: name, delay: delay}
|
||||
}
|
||||
|
||||
func (s *stubPlugin) Name() string { return s.name }
|
||||
func (s *stubPlugin) Start(*sdk.PluginSDK) error { return nil }
|
||||
func (s *stubPlugin) Stop() error {
|
||||
s.stopCount.Add(1)
|
||||
if s.panicOnStop {
|
||||
panic("stub: 故意 panic")
|
||||
}
|
||||
if s.delay > 0 {
|
||||
time.Sleep(s.delay)
|
||||
}
|
||||
s.stopped.Store(true)
|
||||
return nil
|
||||
}
|
||||
|
||||
// StopAll 必须真的并行,否则退回串行 = 关停超时。
|
||||
func TestStopAllStopsInParallel(t *testing.T) {
|
||||
const n = 8
|
||||
const each = 120 * time.Millisecond
|
||||
|
||||
var plugins []sdk.Plugin
|
||||
var stubs []*stubPlugin
|
||||
for i := 0; i < n; i++ {
|
||||
s := newStubPlugin("p", each)
|
||||
_ = i
|
||||
stubs = append(stubs, s)
|
||||
plugins = append(plugins, s)
|
||||
}
|
||||
r := &Registry{instances: plugins}
|
||||
|
||||
start := time.Now()
|
||||
r.StopAll()
|
||||
elapsed := time.Since(start)
|
||||
|
||||
serial := n * each
|
||||
// 串行实现会耗时 serial;留一半余量仍能可靠区分。
|
||||
if elapsed > serial/2 {
|
||||
t.Errorf("StopAll 是串行的:%d 个插件各 %v 用了 %v(串行≈%v,并行应≈%v)",
|
||||
n, each, elapsed, serial, each)
|
||||
}
|
||||
for _, s := range stubs {
|
||||
if !s.stopped.Load() {
|
||||
t.Errorf("插件 %s 没被停掉", s.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 一个插件 panic 不能带崩整个关停,也不能让其它插件停不掉。
|
||||
func TestStopAllSurvivesPanickingPlugin(t *testing.T) {
|
||||
bad := newStubPlugin("bad", 0)
|
||||
bad.panicOnStop = true
|
||||
good := newStubPlugin("good", 0)
|
||||
r := &Registry{instances: []sdk.Plugin{bad, good}}
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
defer func() {
|
||||
if rec := recover(); rec != nil {
|
||||
t.Errorf("插件的 panic 不该冒到关停流程上:%v", rec)
|
||||
}
|
||||
}()
|
||||
r.StopAll()
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("StopAll 卡死(可能死锁)")
|
||||
}
|
||||
if !good.stopped.Load() {
|
||||
t.Error("一个插件 panic 不该让其它插件停不掉")
|
||||
}
|
||||
}
|
||||
|
||||
// 关停期间必须冻结自动重启,否则崩溃判定会在关停中把插件重新拉起
|
||||
// (段已拆而进程还在 → SIGBUS)。
|
||||
func TestStopAllFreezesAutoRestart(t *testing.T) {
|
||||
s := newStubPlugin("p", 0)
|
||||
r := &Registry{instances: []sdk.Plugin{s}}
|
||||
r.StopAll()
|
||||
if !r.shuttingDown.Load() {
|
||||
t.Error("StopAll 之后 shuttingDown 应为 true")
|
||||
}
|
||||
if r.instances != nil {
|
||||
t.Error("StopAll 之后 instances 应被清空")
|
||||
}
|
||||
}
|
||||
|
||||
// 每个插件恰好 Stop 一次:重复调用会二次释放共享段/重复跑 stop handler。
|
||||
func TestStopAllStopsEachPluginExactlyOnce(t *testing.T) {
|
||||
var plugins []sdk.Plugin
|
||||
var stubs []*stubPlugin
|
||||
for i := 0; i < 5; i++ {
|
||||
s := newStubPlugin("p", 0)
|
||||
stubs = append(stubs, s)
|
||||
plugins = append(plugins, s)
|
||||
}
|
||||
r := &Registry{instances: plugins}
|
||||
r.StopAll()
|
||||
for _, s := range stubs {
|
||||
if c := s.stopCount.Load(); c != 1 {
|
||||
t.Errorf("Stop 被调 %d 次,应恰好 1 次", c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// stop handler 必须在 Stop 之前跑完(handler 负责解绑通道等),
|
||||
// 且并行化后这个顺序不能被破坏。
|
||||
//
|
||||
// handler 挂在 PluginSDK 上(RegisterStopHandler),由 runStopHandlers
|
||||
// 通过 r.sdkRefs 取出执行 —— 所以判据必须真的构造一个 PluginSDK,
|
||||
// 否则测的是一条不存在的注册路径。
|
||||
func TestStopAllRunsStopHandlerBeforeStop(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
var order []string
|
||||
|
||||
s := newStubPlugin("p", 0)
|
||||
r := &Registry{
|
||||
instances: []sdk.Plugin{s},
|
||||
sdkRefs: map[string]*sdk.PluginSDK{},
|
||||
}
|
||||
sdkn := sdk.New("p", sdk.SDKConfig{})
|
||||
sdkn.RegisterStopHandler(func() {
|
||||
mu.Lock()
|
||||
order = append(order, "handler")
|
||||
mu.Unlock()
|
||||
})
|
||||
r.sdkRefs["p"] = sdkn
|
||||
|
||||
probe := &orderProbePlugin{stubPlugin: s, onStop: func() {
|
||||
mu.Lock()
|
||||
order = append(order, "stop")
|
||||
mu.Unlock()
|
||||
}}
|
||||
r.instances = []sdk.Plugin{probe}
|
||||
|
||||
r.StopAll()
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(order) != 2 || order[0] != "handler" || order[1] != "stop" {
|
||||
t.Errorf("stop handler 必须先于 Stop 执行,实际顺序 %v", order)
|
||||
}
|
||||
}
|
||||
|
||||
type orderProbePlugin struct {
|
||||
*stubPlugin
|
||||
onStop func()
|
||||
}
|
||||
|
||||
func (o *orderProbePlugin) Stop() error {
|
||||
o.onStop()
|
||||
return o.stubPlugin.Stop()
|
||||
}
|
||||
Reference in New Issue
Block a user