mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-04 00:03:59 +00:00
fix: 插件增量重载 + clawhubadapter ipcGoroutine 优雅退出
- Registry.Reload 改为增量: 对比插件入口文件(plugin.so/main.lua) SHA256, 仅重载有变更的插件, 未变更保持运行, 消除 plgreload 触发全量 StopAll+Load 导致的重复加载与内置插件状态错乱 - Registry 新增 pluginHashes 记录 + pluginEntryHash 辅助 - clawhubadapter.ipcGoroutine 无退出条件导致重载后旧 goroutine 永久残留(线程累积): 增加 stopCh/stopOnce, Stop() 关闭, ipcGoroutine 用 select 监听退出 - 新增 TestRegistryIncrementalReload 验证无变更跳过/变更重载
This commit is contained in:
@ -1,8 +1,12 @@
|
|||||||
package plugin
|
package plugin
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config"
|
||||||
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -202,3 +206,55 @@ func TestPluginManagerInterfaceReloadOne(t *testing.T) {
|
|||||||
// 编译期契约:Registry 必须实现 PluginManager(含 ReloadOne 单插件重载)。
|
// 编译期契约:Registry 必须实现 PluginManager(含 ReloadOne 单插件重载)。
|
||||||
var _ sdk.PluginManager = (*Registry)(nil)
|
var _ sdk.PluginManager = (*Registry)(nil)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRegistryIncrementalReload(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
plgDir := filepath.Join(dir, "plugins")
|
||||||
|
os.MkdirAll(plgDir, 0755)
|
||||||
|
|
||||||
|
// 一个 Lua 插件
|
||||||
|
luaDir := filepath.Join(plgDir, "reloaddemo")
|
||||||
|
os.MkdirAll(luaDir, 0755)
|
||||||
|
os.WriteFile(filepath.Join(luaDir, "plugin.json"), []byte(`{"name":"reloaddemo","entry":"main.lua"}`), 0644)
|
||||||
|
writeLua := func(body string) {
|
||||||
|
os.WriteFile(filepath.Join(luaDir, "main.lua"), []byte(`local plugin = { name = "reloaddemo" }
|
||||||
|
function plugin.start(sdk) sdk.log("info", "`+body+`") end
|
||||||
|
function plugin.stop() end
|
||||||
|
return plugin
|
||||||
|
`), 0644)
|
||||||
|
}
|
||||||
|
writeLua("v1")
|
||||||
|
|
||||||
|
reg := NewRegistry()
|
||||||
|
reg.SetPluginDir(plgDir)
|
||||||
|
reg.SetConfigRegistry(internalConfig.NewConfigRegistry(""))
|
||||||
|
|
||||||
|
|
||||||
|
// 首次 Reload:应加载 1 个
|
||||||
|
msg, err := reg.Reload(plgDir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("first reload: %v", err)
|
||||||
|
}
|
||||||
|
if len(reg.List()) != 1 {
|
||||||
|
t.Fatalf("first reload loaded=%d, want 1 (%s)", len(reg.List()), msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 无变更再 Reload:不应重载(0 changed, 1 unchanged)
|
||||||
|
msg, err = reg.Reload(plgDir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("second reload: %v", err)
|
||||||
|
}
|
||||||
|
if !strings.Contains(msg, "0 plugins") || !strings.Contains(msg, "1 unchanged") {
|
||||||
|
t.Errorf("unchanged reload should skip: %q", msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 修改 main.lua → 应重载该插件
|
||||||
|
writeLua("v2")
|
||||||
|
msg, err = reg.Reload(plgDir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("changed reload: %v", err)
|
||||||
|
}
|
||||||
|
if !strings.Contains(msg, "1 plugins") {
|
||||||
|
t.Errorf("changed reload should reload 1: %q", msg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -1,6 +1,8 @@
|
|||||||
package plugin
|
package plugin
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"crypto/sha256"
|
||||||
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
@ -97,6 +99,10 @@ type Registry struct {
|
|||||||
|
|
||||||
knownDisabled map[string]bool
|
knownDisabled map[string]bool
|
||||||
allowlist map[string]bool
|
allowlist map[string]bool
|
||||||
|
|
||||||
|
// pluginHashes 记录各插件二进制(plugin.so/main.lua)的 SHA256,
|
||||||
|
// 供增量重载(Reload)对比:仅重载有变更的插件,避免全量 StopAll+Load 导致重复加载。
|
||||||
|
pluginHashes map[string]string
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewRegistry() *Registry {
|
func NewRegistry() *Registry {
|
||||||
@ -106,6 +112,7 @@ func NewRegistry() *Registry {
|
|||||||
pluginAutoRestart: make(map[string]bool),
|
pluginAutoRestart: make(map[string]bool),
|
||||||
sdkRefs: make(map[string]*sdk.PluginSDK),
|
sdkRefs: make(map[string]*sdk.PluginSDK),
|
||||||
knownDisabled: make(map[string]bool),
|
knownDisabled: make(map[string]bool),
|
||||||
|
pluginHashes: make(map[string]string),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -345,6 +352,19 @@ func (r *Registry) isDisabled(name string) bool {
|
|||||||
return r.cfgReg.IsPluginDisabled(name)
|
return r.cfgReg.IsPluginDisabled(name)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// pluginEntryHash 计算插件入口文件(plugin.so 或 main.lua)的 SHA256,用于增量重载对比。
|
||||||
|
// 无入口文件(内置纯工厂插件)返回空字符串(始终视为已加载)。
|
||||||
|
func pluginEntryHash(plgDir string) string {
|
||||||
|
for _, candidate := range []string{"plugin.so", "plugin.dll", "main.lua", "SKILL.md"} {
|
||||||
|
path := filepath.Join(plgDir, candidate)
|
||||||
|
if data, err := os.ReadFile(path); err == nil && len(data) > 0 {
|
||||||
|
sum := sha256.Sum256(data)
|
||||||
|
return hex.EncodeToString(sum[:])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
||||||
func (r *Registry) loadOne(plgDir, name string) bool {
|
func (r *Registry) loadOne(plgDir, name string) bool {
|
||||||
if r.isDisabled(name) {
|
if r.isDisabled(name) {
|
||||||
log.Printf("[plugin] %s is disabled, skipping", name)
|
log.Printf("[plugin] %s is disabled, skipping", name)
|
||||||
@ -410,6 +430,11 @@ func (r *Registry) loadOne(plgDir, name string) bool {
|
|||||||
r.pluginAutoRestart[name] = plgSDK.AutoRestart()
|
r.pluginAutoRestart[name] = plgSDK.AutoRestart()
|
||||||
r.sdkRefs[name] = plgSDK
|
r.sdkRefs[name] = plgSDK
|
||||||
r.instances = append(r.instances, plg)
|
r.instances = append(r.instances, plg)
|
||||||
|
if h := pluginEntryHash(plgDir); h != "" {
|
||||||
|
r.pluginHashes[name] = h
|
||||||
|
} else {
|
||||||
|
delete(r.pluginHashes, name)
|
||||||
|
}
|
||||||
r.mu.Unlock()
|
r.mu.Unlock()
|
||||||
log.Printf("[plugin] loaded: %s", name)
|
log.Printf("[plugin] loaded: %s", name)
|
||||||
return true
|
return true
|
||||||
@ -445,11 +470,55 @@ func (r *Registry) StopAll() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r *Registry) Reload(dir string) (string, error) {
|
func (r *Registry) Reload(dir string) (string, error) {
|
||||||
r.StopAll()
|
if dir == "" {
|
||||||
if err := r.Load(dir); err != nil {
|
dir = r.plgDir
|
||||||
|
}
|
||||||
|
// 增量重载:扫描插件目录,对比入口文件 hash,仅 Stop+重载有变更的插件。
|
||||||
|
// 未变更插件保持运行,避免 plgreload 触发全量 StopAll+Load 导致所有插件重复加载
|
||||||
|
// 及内置插件(如 healthcheck)状态机错乱。
|
||||||
|
entries, err := os.ReadDir(dir)
|
||||||
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
return fmt.Sprintf("loaded %d plugins", len(r.instances)), nil
|
changed := 0
|
||||||
|
remaining := 0
|
||||||
|
for _, entry := range entries {
|
||||||
|
if !entry.IsDir() {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
name := entry.Name()
|
||||||
|
if !r.allowlistAllows(name) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
plgDir := filepath.Join(dir, name)
|
||||||
|
h := pluginEntryHash(plgDir)
|
||||||
|
r.mu.RLock()
|
||||||
|
old := r.pluginHashes[name]
|
||||||
|
loaded := r.plugins[name] != nil
|
||||||
|
r.mu.RUnlock()
|
||||||
|
// 无入口文件(纯内置工厂插件)始终视为已加载;
|
||||||
|
// 有变更或首次出现且未加载 → 需要重载。
|
||||||
|
if !loaded {
|
||||||
|
if r.loadOne(plgDir, name) {
|
||||||
|
changed++
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if h == "" {
|
||||||
|
remaining++
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if old != h {
|
||||||
|
if err := r.ReloadOne(name); err != nil {
|
||||||
|
log.Printf("[plugin] reload %s: %v", name, err)
|
||||||
|
} else {
|
||||||
|
changed++
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
remaining++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return fmt.Sprintf("reloaded %d plugins, %d unchanged", changed, remaining), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *Registry) ReloadOne(name string) error {
|
func (r *Registry) ReloadOne(name string) error {
|
||||||
|
|||||||
@ -53,6 +53,9 @@ type Plugin struct {
|
|||||||
sdk *sdk.PluginSDK
|
sdk *sdk.PluginSDK
|
||||||
dispatcher *RegistryDispatcher
|
dispatcher *RegistryDispatcher
|
||||||
httpClient *http.Client
|
httpClient *http.Client
|
||||||
|
|
||||||
|
stopCh chan struct{}
|
||||||
|
stopOnce sync.Once
|
||||||
}
|
}
|
||||||
|
|
||||||
// pluginSingleton 内核单例引用(Start 时设置),供 SendToChannel/ChannelSender 使用
|
// pluginSingleton 内核单例引用(Start 时设置),供 SendToChannel/ChannelSender 使用
|
||||||
@ -65,6 +68,7 @@ func New(name, skillsDir string) *Plugin {
|
|||||||
}
|
}
|
||||||
return &Plugin{
|
return &Plugin{
|
||||||
name: name,
|
name: name,
|
||||||
|
stopCh: make(chan struct{}),
|
||||||
skillsDir: skillsDir,
|
skillsDir: skillsDir,
|
||||||
simulatorDir: sd,
|
simulatorDir: sd,
|
||||||
dispatcher: NewDispatcher(),
|
dispatcher: NewDispatcher(),
|
||||||
@ -300,10 +304,14 @@ func (p *Plugin) ipcGoroutine(s *sdk.PluginSDK) {
|
|||||||
ticker := time.NewTicker(3 * time.Second)
|
ticker := time.NewTicker(3 * time.Second)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
for range ticker.C {
|
for {
|
||||||
p.mu.Lock()
|
select {
|
||||||
simDir := p.simulatorDir
|
case <-p.stopCh:
|
||||||
p.mu.Unlock()
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
p.mu.Lock()
|
||||||
|
simDir := p.simulatorDir
|
||||||
|
p.mu.Unlock()
|
||||||
if simDir == "" {
|
if simDir == "" {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@ -339,6 +347,7 @@ func (p *Plugin) ipcGoroutine(s *sdk.PluginSDK) {
|
|||||||
}
|
}
|
||||||
os.Remove(reloadPath)
|
os.Remove(reloadPath)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1247,6 +1256,10 @@ func (p *Plugin) loadSidecar(s *sdk.PluginSDK, dir, name string) error {
|
|||||||
func (p *Plugin) Stop() error {
|
func (p *Plugin) Stop() error {
|
||||||
p.mu.Lock()
|
p.mu.Lock()
|
||||||
defer p.mu.Unlock()
|
defer p.mu.Unlock()
|
||||||
|
// 关停 ipcGoroutine(stopCh),避免重载后旧 goroutine 残留导致线程累积
|
||||||
|
p.stopOnce.Do(func() {
|
||||||
|
close(p.stopCh)
|
||||||
|
})
|
||||||
for _, sp := range p.sidecars {
|
for _, sp := range p.sidecars {
|
||||||
sp.Close()
|
sp.Close()
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user