From 324181d66318d48e8ad040840b1ca5d60c4bf1e1 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sat, 26 Sep 2026 16:23:56 +0800 Subject: [PATCH] =?UTF-8?q?fix(plugin):=20StopAll=20=E5=B9=B6=E8=A1=8C?= =?UTF-8?q?=E5=81=9C=E6=8F=92=E4=BB=B6=20=E2=80=94=E2=80=94=20=E4=BF=AE?= =?UTF-8?q?=E5=85=B3=E5=81=9C=E5=BF=85=E7=84=B6=E8=B6=85=E6=97=B6=E8=A2=AB?= =?UTF-8?q?=20SIGKILL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 现象 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 包绿。 --- internal/plugin/registry.go | 55 ++++++- internal/plugin/stopall_parallel_test.go | 190 +++++++++++++++++++++++ 2 files changed, 241 insertions(+), 4 deletions(-) create mode 100644 internal/plugin/stopall_parallel_test.go diff --git a/internal/plugin/registry.go b/internal/plugin/registry.go index 8f1ea5c..ed9fb89 100644 --- a/internal/plugin/registry.go +++ b/internal/plugin/registry.go @@ -707,17 +707,40 @@ func (r *Registry) runOnRemoveHandlers(name string) { } } +// stopTarget 是 StopAll 并行化时的一个停止单元：插件本体 + 它的 SDK 句柄。 +type stopTarget struct { + plugin sdk.Plugin + sdkRef *sdk.PluginSDK +} + func (r *Registry) StopAll() { // 关停开始即冻结自动重启：否则「Stop 触发退出 → 崩溃判定 → 重新 spawn」 // 会在内核正在关停时把子进程又拉起来，段已拆而进程还在，直接 SIGBUS。 r.shuttingDown.Store(true) + // 取插件快照后**立即释放 registry 锁**，再并行停。 + // + // 为何必须并行：串行时最坏耗时 = Σ(每个插件) = 5s(plugin.stop 调用) + // + 5s(等退出) + 2s(收割) = 12s；线上有 23 个子进程插件， + // 即 276s，而 systemd 只给 90s ⇒ 关停必然 timed out 然后 SIGKILL。 + // 实测确实每次都超时（线上日志里 23 个插件全退完了， + // 最后那条 "[homed] stopped" 仍打不出来）。 + // + // 为何先释放锁：p.Stop() 会触发 markExited → onExit → ReclaimOwner， + // 那条链要读共享内存段。持着 registry 锁并行跑，若某插件的 onExit + // 回调需要拿 registry 锁（如摘通道），就是自死锁。 r.mu.Lock() - for _, p := range r.instances { - r.runStopHandlers(p.Name()) - if err := p.Stop(); err != nil { - log.Printf("[plugin] stop %s: %v", p.Name(), err) + snapshot := make([]sdk.Plugin, len(r.instances)) + copy(snapshot, r.instances) + // stop handler 挂在 PluginSDK 上（r.sdkRefs），必须**在清空 sdkRefs 之前** + // 把 handler 跑掉 —— 否则下面并行 goroutine 里就找不到它了。 + // runStopHandlers 自己不加锁（调用方持锁），这里正是持锁状态。 + stoppers := make([]stopTarget, 0, len(snapshot)) + for _, p := range snapshot { + if p == nil { + continue } + stoppers = append(stoppers, stopTarget{plugin: p, sdkRef: r.sdkRefs[p.Name()]}) } r.plugins = make(map[string]sdk.Plugin) r.instances = nil @@ -725,6 +748,30 @@ func (r *Registry) StopAll() { r.sdkRefs = make(map[string]*sdk.PluginSDK) r.mu.Unlock() + // 并行停：每个插件一个 goroutine，等全部完成。 + // 单个插件 panic 不带崩整个关停（那会让剩下的插件全停不掉）， + // 也不静默吞掉（留下日志）。 + var wg sync.WaitGroup + for _, t := range stoppers { + wg.Add(1) + go func(tg stopTarget) { + defer wg.Done() + defer func() { + if rec := recover(); rec != nil { + log.Printf("[plugin] stop %s panic: %v", tg.plugin.Name(), rec) + } + }() + // stop handler（解绑通道等）必须先于 Stop：见 runStopHandlers 注释。 + if tg.sdkRef != nil { + tg.sdkRef.RunStopHandlers() + } + if err := tg.plugin.Stop(); err != nil { + log.Printf("[plugin] stop %s: %v", tg.plugin.Name(), err) + } + }(t) + } + wg.Wait() + // 共享段在全部子进程退出后再释放：插件还持有映射时拆段， // 它们下一次访问就是 SIGBUS。在锁外调用：Close 不需 registry 锁， // 而持锁调它会与 onProcCrash 路径（子进程退出回调）产生锁序风险。 diff --git a/internal/plugin/stopall_parallel_test.go b/internal/plugin/stopall_parallel_test.go new file mode 100644 index 0000000..2d3b4c2 --- /dev/null +++ b/internal/plugin/stopall_parallel_test.go @@ -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() +}