mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 17:38:10 +00:00
feat(proc): 子进程通道 —— RPC 协议 + 进程管理(Part 2 核心)
协议面(protocol.go,§3.2 method id 平移为 method 名): - NDJSON 帧,双向复用同一对 stdio;ID>0 需应答,ID==0 为通知(post-and-forget) - 51 个 C ABI method id 全部平移为可读 method 名并标注原编号对照 编号本身扔掉——加能力不用改两边常量表,不再有 47 夹在 7 和 8 之间的痕迹 - case 25(CORE_FREE_STRING) 无对应 method:进程模型下各自 GC,概念消失 - case 23/24(事件订阅) 与 io.setToolBlocks 今日均为空实现「给不了」, 子进程下首次真正可给(§3.8 能力对齐) - 新增 stage.lock/stage.unlock(C ABI 下不存在跨进程锁概念) - StageInvokeParams 不含 StageContext 数据本身——数据在共享段,只带 stage 名 + seq 进程面(process.go,§2.3 保留现有生命周期机制): - Spawn: 启动 + 握手(协议版本不匹配显式拒绝,不半兼容运行) - readLoop: NDJSON 分派应答/插件反向请求,1MB 单帧上限(大 payload 走 arena) - CallContext: ctx 取消时立即返回**且清理 pending 条目** 对比 cgo:超时只让调用方返回,goroutine 永久卡在 C 调用里(现网泄漏 26 次) - Notify: ID=0 不占 pending 表,满足约束 B(流式逐 token 发布不得等待消费者) - markExited: EOF/退出 → 唤醒全部在途调用 → onExit 回调 这是「把 panic 捕获换成进程退出检测」的落点,plugin_health 逻辑完全复用 - Stop: plugin.stop → 宽限期 → 超时 Kill;Kill 后 OS 回收全部资源,零泄漏 - serveRequest 带 panic 隔离:内核 handler panic 不带崩 readLoop 验证(10 项,真实子进程而非 mock,含 -race): - 握手/工具调用/错误上报(插件失败调用方收到 error,非假成功) - 插件反向调用内核(tool.register + settings.get 双向往返) - **崩溃隔离**:插件 panic → 子进程 exit 2,内核存活、收到 onExit、在途调用不挂死 - 优雅停止 / **Kill 卡死插件**(ctx 超时返回 + pending 清零 + 资源回收) - 通知不等应答(100 条 < 1s)/ 50 并发调用应答不串 / 协议版本不匹配拒绝 接口冻结: git diff third_party/homeagent-sdk/sdk/ 为空
This commit is contained in:
499
internal/plugin/proc/process.go
Normal file
499
internal/plugin/proc/process.go
Normal file
@ -0,0 +1,499 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"gitcode.com/JianFeeeee/HomeAgent/internal/meta"
|
||||
)
|
||||
|
||||
// Process 管理一个外部插件子进程:spawn / 双向 JSON-RPC / 优雅停止 / 崩溃检测。
|
||||
//
|
||||
// 设计依据:docs/zh/架构迁移评估.md §4.1 阶段 2、§2.3(保留现有生命周期机制)
|
||||
//
|
||||
// 与 C ABI 路径的关键差异:
|
||||
// - **崩溃隔离**:插件 panic 只让子进程退出,homed 存活(今日 panic 跨 C 栈可带崩内核)
|
||||
// - **真正的取消**:Kill() 后 OS 回收全部资源,零泄漏
|
||||
// (今日 cgo 调用不可抢占,超时后 OS 线程永久占用,现网已泄漏 26 次,§9.3)
|
||||
// - **可同步等真实结果**:RPC 天然可等应答
|
||||
// (今日 cgo 不可嵌套,output_send 只能异步、永远假成功,§9.4)
|
||||
type Process struct {
|
||||
name string
|
||||
bin string
|
||||
dir string
|
||||
|
||||
cmd *exec.Cmd
|
||||
stdin *bufio.Writer
|
||||
stdout io.ReadCloser
|
||||
|
||||
// writeMu 串行化 stdin 写入:NDJSON 帧不能交错,否则对端解析错乱。
|
||||
writeMu sync.Mutex
|
||||
|
||||
// pending 表:请求 ID → 应答通道。
|
||||
mu sync.Mutex
|
||||
nextID uint64
|
||||
pending map[uint64]chan *Response
|
||||
closed bool
|
||||
|
||||
// handler 处理插件反向发起的调用(51 个 core.* method)。
|
||||
handler RequestHandler
|
||||
|
||||
// exited 在 readLoop 检测到 EOF/进程退出后关闭,用于唤醒所有等待者。
|
||||
exited chan struct{}
|
||||
exitOnce sync.Once
|
||||
exitErr atomic.Pointer[error]
|
||||
readerWG sync.WaitGroup
|
||||
readyOnce sync.Once
|
||||
ready chan struct{}
|
||||
|
||||
// onExit 在进程退出时回调(内核用它喂 plugin_health.recordCrash,
|
||||
// 以及 ForceRelease 释放该插件持有的 stage 锁)。
|
||||
onExit func(name string, err error)
|
||||
|
||||
// shmSize 是握手时告知插件的共享段大小(0 表示本插件不用共享段)。
|
||||
shmSize int
|
||||
}
|
||||
|
||||
// RequestHandler 处理插件 → 内核的调用。
|
||||
// 返回值会被序列化为 Response.Result;返回 error 则序列化为 Response.Error。
|
||||
type RequestHandler func(method string, params json.RawMessage) (interface{}, error)
|
||||
|
||||
// Options 是 Spawn 的可选配置。
|
||||
type Options struct {
|
||||
// Dir 是子进程工作目录(通常为插件目录)。
|
||||
Dir string
|
||||
// Env 追加到子进程环境变量。
|
||||
Env []string
|
||||
// ExtraFiles 传给子进程的额外文件描述符(fd 3 起)。
|
||||
// 共享内存段的 memfd 经此传递——子进程 mmap fd 3 即挂载同一段。
|
||||
ExtraFiles []*os.File
|
||||
// ShmSize 是共享段大小,握手时告知插件(与 ExtraFiles[0] 的 memfd 对应)。
|
||||
ShmSize int
|
||||
// Handler 处理插件反向调用。
|
||||
Handler RequestHandler
|
||||
// OnExit 进程退出回调。
|
||||
OnExit func(name string, err error)
|
||||
// HandshakeTimeout 建链超时,默认 10s。
|
||||
HandshakeTimeout time.Duration
|
||||
}
|
||||
|
||||
// 默认超时。
|
||||
const (
|
||||
defaultHandshakeTimeout = 10 * time.Second
|
||||
// stopGracePeriod 是发出 plugin.stop 后等待进程自行退出的时间。
|
||||
// 超时则 Kill——**这是"真正的取消"**,对比 cgo 路径超时后线程永久泄漏。
|
||||
stopGracePeriod = 5 * time.Second
|
||||
)
|
||||
|
||||
// ErrProcessExited 表示子进程已退出,调用无法完成。
|
||||
var ErrProcessExited = errors.New("proc: 插件进程已退出")
|
||||
|
||||
// Spawn 启动插件子进程并完成握手。
|
||||
func Spawn(name, bin string, opts Options) (*Process, error) {
|
||||
if opts.Handler == nil {
|
||||
return nil, fmt.Errorf("proc: %s 缺少 RequestHandler(插件无法回调内核)", name)
|
||||
}
|
||||
timeout := opts.HandshakeTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = defaultHandshakeTimeout
|
||||
}
|
||||
|
||||
cmd := exec.Command(bin)
|
||||
cmd.Dir = opts.Dir
|
||||
// stderr 直通内核日志:插件的 panic 栈、log 输出可直接看到。
|
||||
cmd.Stderr = os.Stderr
|
||||
if len(opts.Env) > 0 {
|
||||
cmd.Env = append(os.Environ(), opts.Env...)
|
||||
}
|
||||
cmd.ExtraFiles = opts.ExtraFiles
|
||||
|
||||
stdinPipe, err := cmd.StdinPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("proc: %s stdin 管道: %w", name, err)
|
||||
}
|
||||
stdoutPipe, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("proc: %s stdout 管道: %w", name, err)
|
||||
}
|
||||
|
||||
p := &Process{
|
||||
name: name,
|
||||
bin: bin,
|
||||
dir: opts.Dir,
|
||||
cmd: cmd,
|
||||
stdin: bufio.NewWriter(stdinPipe),
|
||||
stdout: stdoutPipe,
|
||||
pending: make(map[uint64]chan *Response),
|
||||
handler: opts.Handler,
|
||||
exited: make(chan struct{}),
|
||||
ready: make(chan struct{}),
|
||||
onExit: opts.OnExit,
|
||||
shmSize: opts.ShmSize,
|
||||
}
|
||||
|
||||
if err := cmd.Start(); err != nil {
|
||||
return nil, fmt.Errorf("proc: 启动 %s (%s): %w", name, bin, err)
|
||||
}
|
||||
|
||||
p.readerWG.Add(1)
|
||||
go p.readLoop()
|
||||
|
||||
// 等 readLoop 就绪后再握手,避免应答早于 reader 启动而丢失。
|
||||
<-p.ready
|
||||
|
||||
if err := p.handshake(timeout); err != nil {
|
||||
p.Kill()
|
||||
return nil, err
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// Name 返回插件名。
|
||||
func (p *Process) Name() string { return p.name }
|
||||
|
||||
// PID 返回子进程 PID(用于诊断/日志)。
|
||||
func (p *Process) PID() int {
|
||||
if p.cmd == nil || p.cmd.Process == nil {
|
||||
return 0
|
||||
}
|
||||
return p.cmd.Process.Pid
|
||||
}
|
||||
|
||||
// Exited 返回一个在进程退出时关闭的通道。
|
||||
func (p *Process) Exited() <-chan struct{} { return p.exited }
|
||||
|
||||
// ExitError 返回进程退出原因(正常退出为 nil)。
|
||||
func (p *Process) ExitError() error {
|
||||
if e := p.exitErr.Load(); e != nil {
|
||||
return *e
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Process) handshake(timeout time.Duration) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
raw, err := p.CallContext(ctx, MethodHandshake, HandshakeParams{
|
||||
Protocol: ProtocolVersion,
|
||||
CoreVersion: meta.Version,
|
||||
PluginName: p.name,
|
||||
ShmVersion: shmVersion,
|
||||
ShmSize: p.shmSize,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("proc: %s 握手失败: %w", p.name, err)
|
||||
}
|
||||
var res HandshakeResult
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
return fmt.Errorf("proc: %s 握手应答解析失败: %w", p.name, err)
|
||||
}
|
||||
if res.Protocol != ProtocolVersion {
|
||||
return fmt.Errorf("proc: %s 协议版本不匹配(插件 %d,内核 %d)——请用配套 plugindev 重编",
|
||||
p.name, res.Protocol, ProtocolVersion)
|
||||
}
|
||||
log.Printf("[proc] %s 已建链(pid=%d protocol=%d sdk=%s)",
|
||||
p.name, p.PID(), res.Protocol, res.SDKVersion)
|
||||
return nil
|
||||
}
|
||||
|
||||
// readLoop 读取子进程 stdout 的 NDJSON 帧,分派为「应答」或「插件发起的请求」。
|
||||
//
|
||||
// 参考 clawhubadapter/sidecarProcess 的成熟做法:大 buffer 防长行截断、
|
||||
// pending 表定位应答、退出时唤醒全部等待者。
|
||||
func (p *Process) readLoop() {
|
||||
defer p.readerWG.Done()
|
||||
|
||||
scanner := bufio.NewScanner(bufio.NewReader(p.stdout))
|
||||
// 单帧上限 1MB:控制面帧本应很小(工具结果中位 93B),
|
||||
// 超大 payload 应走共享内存 arena 而非 RPC 帧。
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
|
||||
p.readyOnce.Do(func() { close(p.ready) })
|
||||
|
||||
for scanner.Scan() {
|
||||
line := scanner.Bytes()
|
||||
if len(line) == 0 {
|
||||
continue
|
||||
}
|
||||
// 帧可能是 Response(有 id 无 method)或 Request(有 method)。
|
||||
var probe struct {
|
||||
ID uint64 `json:"id"`
|
||||
Method string `json:"method"`
|
||||
}
|
||||
if err := json.Unmarshal(line, &probe); err != nil {
|
||||
log.Printf("[proc] %s 收到非法 JSON 帧(%d 字节): %v", p.name, len(line), err)
|
||||
continue
|
||||
}
|
||||
|
||||
if probe.Method != "" {
|
||||
// 插件发起的调用:拷贝一份再交给 goroutine(scanner 会复用底层数组)
|
||||
buf := make([]byte, len(line))
|
||||
copy(buf, line)
|
||||
go p.serveRequest(buf)
|
||||
continue
|
||||
}
|
||||
|
||||
var resp Response
|
||||
if err := json.Unmarshal(line, &resp); err != nil {
|
||||
log.Printf("[proc] %s 应答解析失败: %v", p.name, err)
|
||||
continue
|
||||
}
|
||||
p.mu.Lock()
|
||||
ch, ok := p.pending[resp.ID]
|
||||
delete(p.pending, resp.ID)
|
||||
p.mu.Unlock()
|
||||
if !ok {
|
||||
log.Printf("[proc] %s 收到未知 id=%d 的应答(可能已超时)", p.name, resp.ID)
|
||||
continue
|
||||
}
|
||||
ch <- &resp
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
log.Printf("[proc] %s 读取 stdout 出错: %v", p.name, err)
|
||||
}
|
||||
|
||||
// stdout 关闭(EOF)意味着进程结束——2.5ms 内即可感知(实验 6)。
|
||||
p.markExited()
|
||||
}
|
||||
|
||||
// markExited 回收进程、唤醒所有等待者、触发 onExit 回调。
|
||||
//
|
||||
// 这是「把 panic 捕获换成进程退出检测」的落点(§2.3):
|
||||
// plugin_health 的 recordCrash / 冷却 / 自愈 / pendingReloads 全部逻辑复用,
|
||||
// 只是信号源从 recover() 变成进程退出。
|
||||
func (p *Process) markExited() {
|
||||
p.exitOnce.Do(func() {
|
||||
waitErr := p.cmd.Wait()
|
||||
if waitErr != nil {
|
||||
e := fmt.Errorf("插件进程 %s 异常退出: %w", p.name, waitErr)
|
||||
p.exitErr.Store(&e)
|
||||
log.Printf("[proc] %s 退出: %v", p.name, waitErr)
|
||||
} else {
|
||||
log.Printf("[proc] %s 正常退出", p.name)
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
p.closed = true
|
||||
waiters := make([]chan *Response, 0, len(p.pending))
|
||||
for id, ch := range p.pending {
|
||||
waiters = append(waiters, ch)
|
||||
delete(p.pending, id)
|
||||
}
|
||||
p.mu.Unlock()
|
||||
|
||||
// 唤醒所有在途调用,避免调用方挂死到自己的超时
|
||||
for _, ch := range waiters {
|
||||
ch <- &Response{Error: ErrProcessExited.Error()}
|
||||
}
|
||||
|
||||
close(p.exited)
|
||||
if p.onExit != nil {
|
||||
p.onExit(p.name, p.ExitError())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// serveRequest 处理插件反向发起的调用。
|
||||
func (p *Process) serveRequest(line []byte) {
|
||||
var req Request
|
||||
if err := json.Unmarshal(line, &req); err != nil {
|
||||
log.Printf("[proc] %s 请求解析失败: %v", p.name, err)
|
||||
return
|
||||
}
|
||||
|
||||
// panic 隔离:插件的回调参数可能触发内核 handler 的 panic,
|
||||
// 不能让它带崩整个 readLoop(更不能带崩 homed)。
|
||||
var (
|
||||
result interface{}
|
||||
err error
|
||||
)
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
err = fmt.Errorf("内核 handler 处理 %s 时 panic: %v", req.Method, r)
|
||||
log.Printf("[proc] %s: %v", p.name, err)
|
||||
}
|
||||
}()
|
||||
result, err = p.handler(req.Method, req.Params)
|
||||
}()
|
||||
|
||||
// ID==0 是通知,不回应答(§2.4 约束 B:post-and-forget)
|
||||
if req.ID == 0 {
|
||||
if err != nil {
|
||||
log.Printf("[proc] %s 通知 %s 处理失败: %v", p.name, req.Method, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
resp := Response{ID: req.ID}
|
||||
if err != nil {
|
||||
resp.Error = err.Error()
|
||||
} else if result != nil {
|
||||
if b, mErr := json.Marshal(result); mErr == nil {
|
||||
resp.Result = b
|
||||
} else {
|
||||
resp.Error = fmt.Sprintf("结果序列化失败: %v", mErr)
|
||||
}
|
||||
}
|
||||
if wErr := p.writeFrame(&resp); wErr != nil {
|
||||
log.Printf("[proc] %s 回写应答失败: %v", p.name, wErr)
|
||||
}
|
||||
}
|
||||
|
||||
// writeFrame 序列化并写入一帧(串行化,NDJSON 不能交错)。
|
||||
func (p *Process) writeFrame(v interface{}) error {
|
||||
b, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.writeMu.Lock()
|
||||
defer p.writeMu.Unlock()
|
||||
if _, err := p.stdin.Write(b); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := p.stdin.WriteByte('\n'); err != nil {
|
||||
return err
|
||||
}
|
||||
return p.stdin.Flush()
|
||||
}
|
||||
|
||||
// Call 发起 RPC 并等待应答(无超时上限,由调用方 context 控制)。
|
||||
func (p *Process) Call(method string, params interface{}) (json.RawMessage, error) {
|
||||
return p.CallContext(context.Background(), method, params)
|
||||
}
|
||||
|
||||
// CallContext 发起 RPC 并等待应答,受 ctx 取消/超时控制。
|
||||
//
|
||||
// **ctx 取消时调用方立即返回,且 pending 条目被清理**——
|
||||
// 对比 cgo 路径:超时只让调用方返回,goroutine 仍永久卡在 C 调用里(§9.3)。
|
||||
// 这里子进程若真卡住,上层可 Kill(),OS 回收全部资源。
|
||||
func (p *Process) CallContext(ctx context.Context, method string, params interface{}) (json.RawMessage, error) {
|
||||
var raw json.RawMessage
|
||||
if params != nil {
|
||||
b, err := json.Marshal(params)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("proc: %s 序列化 %s 参数: %w", p.name, method, err)
|
||||
}
|
||||
raw = b
|
||||
}
|
||||
|
||||
ch := make(chan *Response, 1)
|
||||
|
||||
p.mu.Lock()
|
||||
if p.closed {
|
||||
p.mu.Unlock()
|
||||
return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ErrProcessExited)
|
||||
}
|
||||
p.nextID++
|
||||
id := p.nextID
|
||||
p.pending[id] = ch
|
||||
p.mu.Unlock()
|
||||
|
||||
if err := p.writeFrame(&Request{ID: id, Method: method, Params: raw}); err != nil {
|
||||
p.mu.Lock()
|
||||
delete(p.pending, id)
|
||||
p.mu.Unlock()
|
||||
return nil, fmt.Errorf("proc: %s 发送 %s: %w", p.name, method, err)
|
||||
}
|
||||
|
||||
select {
|
||||
case resp := <-ch:
|
||||
if resp.Error != "" {
|
||||
return nil, fmt.Errorf("proc: %s.%s: %s", p.name, method, resp.Error)
|
||||
}
|
||||
return resp.Result, nil
|
||||
case <-ctx.Done():
|
||||
p.mu.Lock()
|
||||
delete(p.pending, id)
|
||||
p.mu.Unlock()
|
||||
return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ctx.Err())
|
||||
case <-p.exited:
|
||||
return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ErrProcessExited)
|
||||
}
|
||||
}
|
||||
|
||||
// Notify 发送不需要应答的通知(ID=0,fire-and-forget)。
|
||||
//
|
||||
// 用于事件投递等路径:内核发通知**绝不等待消费者**(§2.4 约束 B——
|
||||
// 流式输出逐 token 发布,任何等待都会造成卡顿)。
|
||||
func (p *Process) Notify(method string, params interface{}) error {
|
||||
var raw json.RawMessage
|
||||
if params != nil {
|
||||
b, err := json.Marshal(params)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
raw = b
|
||||
}
|
||||
p.mu.Lock()
|
||||
closed := p.closed
|
||||
p.mu.Unlock()
|
||||
if closed {
|
||||
return ErrProcessExited
|
||||
}
|
||||
return p.writeFrame(&Request{Method: method, Params: raw})
|
||||
}
|
||||
|
||||
// Stop 优雅停止:发 plugin.stop → 等宽限期 → 超时则 Kill。
|
||||
//
|
||||
// 插件侧收到 plugin.stop 后应先跑 RunStopHandlers 再 Stop(),
|
||||
// 与 C ABI 路径的停止链路语义一致(§2.3 已验证被正确调用)。
|
||||
func (p *Process) Stop() error {
|
||||
select {
|
||||
case <-p.exited:
|
||||
return nil // 已经退出
|
||||
default:
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stopGracePeriod)
|
||||
defer cancel()
|
||||
if _, err := p.CallContext(ctx, MethodPluginStop, nil); err != nil {
|
||||
// 停止调用失败不影响后续 Kill——插件可能已经崩了
|
||||
if !errors.Is(err, ErrProcessExited) {
|
||||
log.Printf("[proc] %s plugin.stop 失败(将强制结束): %v", p.name, err)
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case <-p.exited:
|
||||
return nil
|
||||
case <-time.After(stopGracePeriod):
|
||||
log.Printf("[proc] %s 宽限期内未退出,强制结束", p.name)
|
||||
return p.Kill()
|
||||
}
|
||||
}
|
||||
|
||||
// Kill 强制结束子进程并回收资源。
|
||||
//
|
||||
// **这是 C ABI 路径拿不到的能力**:cgo 调用不可被 Go runtime 抢占或取消,
|
||||
// 超时后该 OS 线程永久占用(实验 14 实测 20 次调用线性泄漏 +18 线程)。
|
||||
// 子进程模型下 Kill 后 OS 回收全部资源,零泄漏。
|
||||
func (p *Process) Kill() error {
|
||||
if p.cmd == nil || p.cmd.Process == nil {
|
||||
return nil
|
||||
}
|
||||
err := p.cmd.Process.Kill()
|
||||
// 等 readLoop 观察到 EOF 并完成 Wait/清理
|
||||
select {
|
||||
case <-p.exited:
|
||||
case <-time.After(2 * time.Second):
|
||||
p.markExited() // 兜底:极端情况下强制走清理
|
||||
}
|
||||
p.readerWG.Wait()
|
||||
if err != nil && !errors.Is(err, os.ErrProcessDone) {
|
||||
return fmt.Errorf("proc: 结束 %s: %w", p.name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
334
internal/plugin/proc/process_test.go
Normal file
334
internal/plugin/proc/process_test.go
Normal file
@ -0,0 +1,334 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Process 的测试用真实子进程(go build 出的小二进制),而非 mock:
|
||||
// 崩溃隔离、EOF 感知、Kill 回收这些性质只有真进程才能验证——
|
||||
// 它们恰是迁移相对 C ABI 的核心收益(§9.5)。
|
||||
|
||||
// buildTestPlugin 编译 testdata 下的假插件,返回二进制路径。
|
||||
func buildTestPlugin(t *testing.T, srcName string) string {
|
||||
t.Helper()
|
||||
if _, err := exec.LookPath("go"); err != nil {
|
||||
t.Skip("环境无 go 工具链,跳过子进程测试")
|
||||
}
|
||||
|
||||
src := filepath.Join("testdata", srcName)
|
||||
if _, err := os.Stat(src); err != nil {
|
||||
t.Fatalf("测试插件源码缺失 %s: %v", src, err)
|
||||
}
|
||||
|
||||
bin := filepath.Join(t.TempDir(), strings.TrimSuffix(srcName, ".go"))
|
||||
cmd := exec.Command("go", "build", "-o", bin, src)
|
||||
cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
|
||||
if out, err := cmd.CombinedOutput(); err != nil {
|
||||
t.Fatalf("编译测试插件 %s 失败: %v\n%s", srcName, err, out)
|
||||
}
|
||||
return bin
|
||||
}
|
||||
|
||||
// noopHandler 是最简的内核侧 handler(测试中不需要真实 core.* 能力)。
|
||||
func noopHandler(method string, params json.RawMessage) (interface{}, error) {
|
||||
return nil, fmt.Errorf("测试环境未实现 %s", method)
|
||||
}
|
||||
|
||||
func TestProcess_SpawnHandshakeAndToolInvoke(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "echoplugin.go")
|
||||
|
||||
p, err := Spawn("echo", bin, Options{Handler: noopHandler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
defer p.Kill()
|
||||
|
||||
if p.PID() == 0 {
|
||||
t.Error("PID 应非零")
|
||||
}
|
||||
|
||||
raw, err := p.Call(MethodToolInvoke, ToolInvokeParams{
|
||||
Name: "echo_tool",
|
||||
Args: map[string]interface{}{"text": "你好"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("tool.invoke: %v", err)
|
||||
}
|
||||
var res ToolInvokeResult
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
t.Fatalf("解析应答: %v", err)
|
||||
}
|
||||
if res.Result != "你好" {
|
||||
t.Fatalf("工具应回显 '你好',实际 %v", res.Result)
|
||||
}
|
||||
}
|
||||
|
||||
// 插件返回错误时调用方必须收到 error —— 对比 C ABI 路径的 output_send
|
||||
// 永远返回成功(§9.4,现网 2 次消息发不出而模型以为成功)。
|
||||
func TestProcess_PluginErrorIsReported(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "echoplugin.go")
|
||||
p, err := Spawn("echo", bin, Options{Handler: noopHandler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
defer p.Kill()
|
||||
|
||||
_, err = p.Call(MethodToolInvoke, ToolInvokeParams{Name: "fail_tool"})
|
||||
if err == nil {
|
||||
t.Fatal("插件返回错误时调用方应收到 error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "故意失败") {
|
||||
t.Errorf("错误信息应透传插件的原因,实际: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 插件反向调用内核(51 个 core.* method 的机制验证)。
|
||||
func TestProcess_PluginCallsBackIntoKernel(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "callbackplugin.go")
|
||||
|
||||
var (
|
||||
mu sync.Mutex
|
||||
gotCall []string
|
||||
)
|
||||
handler := func(method string, params json.RawMessage) (interface{}, error) {
|
||||
mu.Lock()
|
||||
gotCall = append(gotCall, method)
|
||||
mu.Unlock()
|
||||
switch method {
|
||||
case MethodSettingsGet:
|
||||
return map[string]interface{}{"value": "配置值"}, nil
|
||||
case MethodToolRegister:
|
||||
return nil, nil
|
||||
}
|
||||
return nil, fmt.Errorf("未实现 %s", method)
|
||||
}
|
||||
|
||||
p, err := Spawn("cb", bin, Options{Handler: handler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
defer p.Kill()
|
||||
|
||||
// plugin.start 期间插件会回调 tool.register + settings.get
|
||||
if _, err := p.Call(MethodPluginStart, nil); err != nil {
|
||||
t.Fatalf("plugin.start: %v", err)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(gotCall) < 2 {
|
||||
t.Fatalf("内核应收到插件的反向调用,实际 %v", gotCall)
|
||||
}
|
||||
hasRegister, hasSettings := false, false
|
||||
for _, m := range gotCall {
|
||||
if m == MethodToolRegister {
|
||||
hasRegister = true
|
||||
}
|
||||
if m == MethodSettingsGet {
|
||||
hasSettings = true
|
||||
}
|
||||
}
|
||||
if !hasRegister || !hasSettings {
|
||||
t.Errorf("应收到 tool.register 与 settings.get,实际 %v", gotCall)
|
||||
}
|
||||
}
|
||||
|
||||
// 崩溃隔离:插件 panic 只让子进程退出,内核存活并收到 onExit(§9.5 表格第 2 行)。
|
||||
// C ABI 路径下 panic 跨 C 栈,recover 兜不住会带崩整个 homed(§1.4)。
|
||||
func TestProcess_CrashIsolationAndExitDetection(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "crashplugin.go")
|
||||
|
||||
exitCh := make(chan error, 1)
|
||||
p, err := Spawn("crash", bin, Options{
|
||||
Handler: noopHandler,
|
||||
OnExit: func(name string, err error) { exitCh <- err },
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
|
||||
// 触发插件 panic
|
||||
_, callErr := p.Call(MethodToolInvoke, ToolInvokeParams{Name: "boom"})
|
||||
if callErr == nil {
|
||||
t.Error("插件崩溃时在途调用应返回错误,而非挂死")
|
||||
}
|
||||
|
||||
select {
|
||||
case exitErr := <-exitCh:
|
||||
if exitErr == nil {
|
||||
t.Error("panic 退出应报告非 nil 错误(供 recordCrash 使用)")
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("未在 5s 内检测到进程退出(EOF 感知失效)")
|
||||
}
|
||||
|
||||
select {
|
||||
case <-p.Exited():
|
||||
case <-time.After(time.Second):
|
||||
t.Error("Exited() 通道应已关闭")
|
||||
}
|
||||
|
||||
// 进程已退出后继续调用应立即失败,不能挂死
|
||||
if _, err := p.Call(MethodToolInvoke, ToolInvokeParams{Name: "echo_tool"}); !errors.Is(err, ErrProcessExited) {
|
||||
t.Errorf("退出后调用应返回 ErrProcessExited,实际 %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 优雅停止:plugin.stop 后进程自行退出。
|
||||
func TestProcess_GracefulStop(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "echoplugin.go")
|
||||
p, err := Spawn("echo", bin, Options{Handler: noopHandler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
|
||||
if err := p.Stop(); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-p.Exited():
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("Stop 后进程应退出")
|
||||
}
|
||||
if err := p.ExitError(); err != nil {
|
||||
t.Errorf("优雅停止应无错误退出,实际 %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// **真正的取消**:卡死的插件可被 Kill 回收(§9.3 对照 cgo 超时线程永久泄漏)。
|
||||
func TestProcess_KillHungPlugin(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "hangplugin.go")
|
||||
p, err := Spawn("hang", bin, Options{Handler: noopHandler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
|
||||
// 调用会卡住,用 context 超时返回(调用方不被拖死)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
|
||||
defer cancel()
|
||||
_, err = p.CallContext(ctx, MethodToolInvoke, ToolInvokeParams{Name: "hang_tool"})
|
||||
if err == nil {
|
||||
t.Fatal("卡死的调用应因 ctx 超时返回")
|
||||
}
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Errorf("应为 DeadlineExceeded,实际 %v", err)
|
||||
}
|
||||
|
||||
// pending 条目必须已清理(不泄漏)
|
||||
p.mu.Lock()
|
||||
pendingCount := len(p.pending)
|
||||
p.mu.Unlock()
|
||||
if pendingCount != 0 {
|
||||
t.Errorf("超时后 pending 表应清空,实际残留 %d 条", pendingCount)
|
||||
}
|
||||
|
||||
// Kill 真正回收资源
|
||||
if err := p.Kill(); err != nil {
|
||||
t.Fatalf("Kill: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-p.Exited():
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("Kill 后进程应退出")
|
||||
}
|
||||
}
|
||||
|
||||
// 通知(ID=0)不等应答——事件投递路径必须 post-and-forget(§2.4 约束 B)。
|
||||
func TestProcess_NotifyDoesNotWait(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "echoplugin.go")
|
||||
p, err := Spawn("echo", bin, Options{Handler: noopHandler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
defer p.Kill()
|
||||
|
||||
start := time.Now()
|
||||
for i := 0; i < 100; i++ {
|
||||
if err := p.Notify("event.deliver", map[string]interface{}{"seq": i}); err != nil {
|
||||
t.Fatalf("Notify: %v", err)
|
||||
}
|
||||
}
|
||||
elapsed := time.Since(start)
|
||||
// 100 条通知若每条都等应答,至少要 100 个往返;post-and-forget 应远快于此
|
||||
if elapsed > time.Second {
|
||||
t.Errorf("100 条通知耗时 %v,疑似在等应答(应 post-and-forget)", elapsed)
|
||||
}
|
||||
|
||||
// 通知不占 pending 表
|
||||
p.mu.Lock()
|
||||
pendingCount := len(p.pending)
|
||||
p.mu.Unlock()
|
||||
if pendingCount != 0 {
|
||||
t.Errorf("通知不应占用 pending 表,实际 %d 条", pendingCount)
|
||||
}
|
||||
}
|
||||
|
||||
// 并发调用:pending 表按 ID 正确路由,应答不串。
|
||||
func TestProcess_ConcurrentCallsRouteCorrectly(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "echoplugin.go")
|
||||
p, err := Spawn("echo", bin, Options{Handler: noopHandler})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
defer p.Kill()
|
||||
|
||||
const n = 50
|
||||
var wg sync.WaitGroup
|
||||
errs := make(chan error, n)
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
want := fmt.Sprintf("msg-%d", i)
|
||||
raw, err := p.Call(MethodToolInvoke, ToolInvokeParams{
|
||||
Name: "echo_tool",
|
||||
Args: map[string]interface{}{"text": want},
|
||||
})
|
||||
if err != nil {
|
||||
errs <- err
|
||||
return
|
||||
}
|
||||
var res ToolInvokeResult
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
errs <- err
|
||||
return
|
||||
}
|
||||
if res.Result != want {
|
||||
errs <- fmt.Errorf("应答串了:期望 %q,实际 %v", want, res.Result)
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
close(errs)
|
||||
for err := range errs {
|
||||
t.Errorf("并发调用失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 协议版本不匹配必须显式拒绝,不能半兼容运行。
|
||||
func TestProcess_ProtocolMismatchRejected(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "badprotoplugin.go")
|
||||
_, err := Spawn("badproto", bin, Options{Handler: noopHandler})
|
||||
if err == nil {
|
||||
t.Fatal("协议版本不匹配应拒绝建链")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "协议版本不匹配") {
|
||||
t.Errorf("错误应说明版本不匹配,实际: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcess_SpawnRequiresHandler(t *testing.T) {
|
||||
if _, err := Spawn("x", "/bin/true", Options{}); err == nil {
|
||||
t.Fatal("缺少 Handler 应报错(插件无法回调内核)")
|
||||
}
|
||||
}
|
||||
215
internal/plugin/proc/protocol.go
Normal file
215
internal/plugin/proc/protocol.go
Normal file
@ -0,0 +1,215 @@
|
||||
package proc
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
// RPC 协议定义:控制面(§3.2 method id 平移为 method 名)。
|
||||
//
|
||||
// 帧格式:**换行分隔的 JSON**(NDJSON),双向复用同一对 stdio 管道。
|
||||
// 内核 → 插件 stdin :请求 / 响应
|
||||
// 插件 → 内核 stdout:请求 / 响应
|
||||
//
|
||||
// 为什么不用 length-prefixed 二进制帧:工具调用结果中位数仅 93B(§2.5),
|
||||
// JSON 序列化 3-8 µs 对比 LLM 单轮 2-8 秒占 0.0001%,可读性与可调试性更值。
|
||||
// 大 payload(多媒体二进制)走共享内存 arena,不进 RPC 帧(§3.3 实验 10:18-22x)。
|
||||
|
||||
// 协议版本:与共享段版本独立演进。
|
||||
// 插件握手时上报,内核校验——不匹配显式拒绝,避免半兼容导致的诡异行为。
|
||||
const ProtocolVersion = 1
|
||||
|
||||
// Direction 无需显式字段:靠 Method 是否为空区分请求与响应
|
||||
// (与 clawhubadapter/sidecar 的成熟做法一致)。
|
||||
|
||||
// Request 是一次 RPC 调用。
|
||||
//
|
||||
// ID 语义:
|
||||
// - ID > 0 :需要响应,调用方在 pending 表等待
|
||||
// - ID == 0 :通知(fire-and-forget),被调方不得回响应
|
||||
//
|
||||
// 通知用于事件投递等不关心结果的路径(§2.4 约束 B:内核发通知绝不等待消费者)。
|
||||
type Request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
// Response 是对 Request 的应答。Error 非空表示失败。
|
||||
type Response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result json.RawMessage `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// ---- kernel → plugin(内核调用插件,对应今日 7 个 //export)----
|
||||
const (
|
||||
// MethodPluginInit 传插件名与配置,插件构造实例但不启动。
|
||||
MethodPluginInit = "plugin.init"
|
||||
// MethodPluginStart 插件注册工具/阶段/通道(其间会反向发起大量 core.* 调用)。
|
||||
MethodPluginStart = "plugin.start"
|
||||
// MethodPluginStop 优雅停止:插件侧先跑 RunStopHandlers 再 Stop()。
|
||||
MethodPluginStop = "plugin.stop"
|
||||
// MethodToolInvoke 执行插件工具。
|
||||
MethodToolInvoke = "tool.invoke"
|
||||
// MethodStageInvoke 执行阶段处理器。数据经共享段传递,参数只带阶段名与段世代号。
|
||||
MethodStageInvoke = "stage.invoke"
|
||||
// MethodOutputInvoke 经插件输出通道发送。
|
||||
MethodOutputInvoke = "output.invoke"
|
||||
// MethodHandshake 建链首帧:交换协议版本、SDK 版本、共享段规格。
|
||||
MethodHandshake = "handshake"
|
||||
)
|
||||
|
||||
// ---- plugin → kernel(51 个 method id 平移,§3.2)----
|
||||
//
|
||||
// 编号本身扔掉:不再维护"下一个可用 id 是 52",加能力不用改两边常量表,
|
||||
// 也不再出现 47 夹在 7 和 8 之间的历史痕迹。
|
||||
const (
|
||||
// 注册面(原 case 1/2/3/4/46)
|
||||
MethodToolRegister = "tool.register" // 1 CORE_REGISTER_TOOL
|
||||
MethodStageRegister = "stage.register" // 2 CORE_REGISTER_STAGE
|
||||
MethodOutputRegister = "output.register" // 3 CORE_REGISTER_OUTPUT_CH
|
||||
MethodAPIRegister = "api.register" // 4 CORE_REGISTER_PLUGIN_API
|
||||
MethodInputRegister = "input.register" // 46 CORE_REGISTER_INPUT_CH
|
||||
|
||||
// IO 注入(原 case 5/6/7/47)
|
||||
MethodIOInjectText = "io.injectText" // 5 CORE_INJECT_TEXT
|
||||
MethodIOInjectInterrupt = "io.injectInterrupt" // 6 CORE_INJECT_INTERRUPT_TEXT
|
||||
MethodIOInjectTextNoMem = "io.injectTextNoMem" // 7 CORE_INJECT_TEXT_NO_MEMORY
|
||||
MethodIOInjectSync = "io.injectInputSync" // 47 CORE_INJECT_INPUT_SYNC
|
||||
// MethodIOSetToolBlocks 多模态注入——今日 C ABI 侧是空实现(§1.4),
|
||||
// 子进程下二进制落 arena、描述符回传,首次真正可用。
|
||||
MethodIOSetToolBlocks = "io.setToolBlocks"
|
||||
|
||||
// 生命周期(原 case 8)
|
||||
MethodLifecycleAutoRestart = "lifecycle.autoRestart" // 8 CORE_SET_AUTO_RESTART
|
||||
|
||||
// 图记忆(原 case 9/10/11/12/13)
|
||||
MethodMemoryRecall = "memory.recall" // 9
|
||||
MethodMemoryCommit = "memory.commit" // 10
|
||||
MethodMemoryIntrospect = "memory.introspect" // 11
|
||||
MethodMemoryMerge = "memory.merge" // 12
|
||||
MethodMemoryPurge = "memory.purge" // 13
|
||||
|
||||
// 文档记忆(原 case 14/32/33/34)
|
||||
MethodDocQuery = "doc.query" // 14
|
||||
MethodDocInsert = "doc.insert" // 32
|
||||
MethodDocRemove = "doc.remove" // 33
|
||||
MethodDocStats = "doc.stats" // 34
|
||||
|
||||
// 知识库(原 case 15/35/36)
|
||||
MethodKnowledgeSearch = "knowledge.search" // 15
|
||||
MethodKnowledgeAdd = "knowledge.add" // 35
|
||||
MethodKnowledgeList = "knowledge.list" // 36
|
||||
|
||||
// 文本记忆(原 case 41)
|
||||
MethodTextMemoryAppend = "textmemory.append" // 41
|
||||
|
||||
// 设置(原 case 16/17/18/26/27/28/29/30/31/42/43/44/45/51)
|
||||
MethodSettingsGet = "settings.get" // 16
|
||||
MethodSettingsSet = "settings.set" // 17
|
||||
MethodSettingsRegisterDef = "settings.registerDef" // 18
|
||||
MethodSettingsGetCore = "settings.getCore" // 26
|
||||
MethodSettingsSetCore = "settings.setCore" // 27
|
||||
MethodSettingsListCore = "settings.listCore" // 28
|
||||
MethodSettingsGetPlugin = "settings.getPlugin" // 29
|
||||
MethodSettingsSetPlugin = "settings.setPlugin" // 30
|
||||
MethodSettingsListPlugin = "settings.listPlugin" // 31
|
||||
MethodSettingsList = "settings.list" // 42
|
||||
MethodSettingsDefs = "settings.defs" // 43
|
||||
MethodSettingsDump = "settings.dump" // 44
|
||||
MethodSettingsPlugins = "settings.plugins" // 45
|
||||
MethodSettingsDataDir = "settings.dataDir" // 51
|
||||
|
||||
// LLM 源(原 case 19/20/37)
|
||||
MethodLLMListSources = "llm.listSources" // 19
|
||||
MethodLLMSetSource = "llm.setSource" // 20
|
||||
MethodLLMCurrentSource = "llm.currentSource" // 37
|
||||
|
||||
// 社交图(只读,原 case 21/22/38/39/40)
|
||||
MethodSocialGetPerson = "social.getPerson" // 21
|
||||
MethodSocialGetNetwork = "social.getNetwork" // 22
|
||||
MethodSocialGetTrait = "social.getTrait" // 38
|
||||
MethodSocialGetRelation = "social.getRelations" // 39
|
||||
MethodSocialListPersons = "social.listPersons" // 40
|
||||
|
||||
// 事件(原 case 23/24 —— 今日均为空实现「给不了」,
|
||||
// 子进程下经事件环 + eventfd 首次真正可用,见 §3.6/§3.8)
|
||||
MethodEventsSubscribe = "events.subscribe" // 23
|
||||
MethodEventsUnsubscribe = "events.unsubscribe" // 24
|
||||
|
||||
// 插件管理(原 case 48/49/50)
|
||||
MethodPluginReloadOne = "plugin.reloadOne" // 48
|
||||
MethodPluginListLoaded = "plugin.listLoaded" // 49
|
||||
MethodPluginIsDisabled = "plugin.isDisabled" // 50
|
||||
|
||||
// 共享段锁仲裁(新增,无对应 method id —— C ABI 下不存在跨进程锁概念)
|
||||
MethodStageLock = "stage.lock"
|
||||
MethodStageUnlock = "stage.unlock"
|
||||
)
|
||||
|
||||
// 原 case 25(CORE_FREE_STRING)无对应 RPC method:
|
||||
// C ABI 下需要显式释放跨边界字符串,进程模型下由各自 GC 管理,概念消失。
|
||||
|
||||
// HandshakeParams 是内核 → 插件的建链首帧:告知内核侧规格。
|
||||
type HandshakeParams struct {
|
||||
Protocol int `json:"protocol"` // 内核支持的协议版本
|
||||
CoreVersion string `json:"core_version"` // 内核版本(诊断用)
|
||||
PluginName string `json:"plugin_name"` // 内核分配的插件名
|
||||
// ShmVersion 让插件确认共享段布局一致;不匹配时插件应拒绝启动而非错读。
|
||||
ShmVersion uint32 `json:"shm_version"`
|
||||
// ShmSize 是内核分配的共享段大小,插件据此 mmap(段本身经 fd 3 传入)。
|
||||
ShmSize int `json:"shm_size"`
|
||||
}
|
||||
|
||||
// HandshakeResult 是插件 → 内核的建链应答:上报自身信息。
|
||||
type HandshakeResult struct {
|
||||
Protocol int `json:"protocol"` // 必须等于 ProtocolVersion
|
||||
SDKVersion string `json:"sdk_version"` // 插件编译时链接的公开 SDK 版本
|
||||
PluginName string `json:"plugin_name"`
|
||||
PID int `json:"pid"`
|
||||
}
|
||||
|
||||
// StageInvokeParams 是 stage.invoke 的参数。
|
||||
//
|
||||
// **注意:不含 StageContext 数据本身**——数据在共享段,此处只带定位信息。
|
||||
// 这是共享内存数据面的意义:并发改写同一份状态,而非各持副本
|
||||
// (副本模型实测 35.8~36.8% lost update,§8.4)。
|
||||
type StageInvokeParams struct {
|
||||
Stage string `json:"stage"`
|
||||
// Seq 是内核写入共享段后的世代号,插件读到的 seq 应 >= 此值。
|
||||
Seq uint64 `json:"seq"`
|
||||
}
|
||||
|
||||
// StageInvokeResult 是插件执行 stage 后的应答。
|
||||
type StageInvokeResult struct {
|
||||
// DirtyFields 是插件实际写回共享段的字段数,0 表示只读插件。
|
||||
// 内核据此判断是否需要重读共享段,也用于诊断"谁改了什么"。
|
||||
DirtyFields int `json:"dirty_fields"`
|
||||
// Seq 是插件写回后的世代号。
|
||||
Seq uint64 `json:"seq"`
|
||||
}
|
||||
|
||||
// ToolInvokeParams / ToolInvokeResult:工具调用(原 go_invoke_tool)。
|
||||
type ToolInvokeParams struct {
|
||||
Name string `json:"name"`
|
||||
Args map[string]interface{} `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
type ToolInvokeResult struct {
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
}
|
||||
|
||||
// OutputInvokeParams:输出通道发送(原 go_invoke_output)。
|
||||
//
|
||||
// 与 C ABI 路径的关键差异:**可同步等待真实结果**。
|
||||
// C ABI 下因 cgo 不可嵌套,只能异步 fire-and-forget,导致 output_send
|
||||
// 永远返回成功(§9.4,现网 2 次消息发不出而模型以为成功)。
|
||||
// 进程模型下 RPC 天然可等应答,该缺陷从根上消失。
|
||||
type OutputInvokeParams struct {
|
||||
Channel string `json:"channel"`
|
||||
Args map[string]interface{} `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
// PluginInitParams:插件构造参数(原 case init_plugin)。
|
||||
type PluginInitParams struct {
|
||||
Name string `json:"name"`
|
||||
Config map[string]interface{} `json:"config,omitempty"`
|
||||
}
|
||||
52
internal/plugin/proc/testdata/badprotoplugin.go
vendored
Normal file
52
internal/plugin/proc/testdata/badprotoplugin.go
vendored
Normal file
@ -0,0 +1,52 @@
|
||||
//go:build ignore
|
||||
|
||||
// badprotoplugin 上报错误的协议版本,验证内核显式拒绝而非半兼容运行。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"os"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
out := bufio.NewWriter(os.Stdout)
|
||||
send := func(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
}
|
||||
|
||||
for in.Scan() {
|
||||
var req request
|
||||
if err := json.Unmarshal(in.Bytes(), &req); err != nil {
|
||||
continue
|
||||
}
|
||||
if req.Method == "handshake" {
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 999, // 故意不匹配
|
||||
"sdk_version": "ancient",
|
||||
"plugin_name": "badproto",
|
||||
"pid": os.Getpid(),
|
||||
}})
|
||||
continue
|
||||
}
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
120
internal/plugin/proc/testdata/callbackplugin.go
vendored
Normal file
120
internal/plugin/proc/testdata/callbackplugin.go
vendored
Normal file
@ -0,0 +1,120 @@
|
||||
//go:build ignore
|
||||
|
||||
// callbackplugin 验证插件 → 内核的反向调用(51 个 core.* method 的机制)。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
var (
|
||||
out = bufio.NewWriter(os.Stdout)
|
||||
writeMu sync.Mutex
|
||||
nextID uint64
|
||||
pending = map[uint64]chan json.RawMessage{}
|
||||
pendMu sync.Mutex
|
||||
)
|
||||
|
||||
func send(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
writeMu.Lock()
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
writeMu.Unlock()
|
||||
}
|
||||
|
||||
// callKernel 反向调用内核并等待应答。
|
||||
func callKernel(method string, params interface{}) json.RawMessage {
|
||||
pendMu.Lock()
|
||||
nextID++
|
||||
id := nextID
|
||||
ch := make(chan json.RawMessage, 1)
|
||||
pending[id] = ch
|
||||
pendMu.Unlock()
|
||||
|
||||
var raw json.RawMessage
|
||||
if params != nil {
|
||||
b, _ := json.Marshal(params)
|
||||
raw = b
|
||||
}
|
||||
send(request{ID: id, Method: method, Params: raw})
|
||||
return <-ch
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
in.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
|
||||
for in.Scan() {
|
||||
line := make([]byte, len(in.Bytes()))
|
||||
copy(line, in.Bytes())
|
||||
|
||||
var probe struct {
|
||||
ID uint64 `json:"id"`
|
||||
Method string `json:"method"`
|
||||
}
|
||||
if err := json.Unmarshal(line, &probe); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// 内核对我们反向调用的应答
|
||||
if probe.Method == "" {
|
||||
var resp struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result json.RawMessage `json:"result"`
|
||||
}
|
||||
json.Unmarshal(line, &resp)
|
||||
pendMu.Lock()
|
||||
ch, ok := pending[resp.ID]
|
||||
delete(pending, resp.ID)
|
||||
pendMu.Unlock()
|
||||
if ok {
|
||||
ch <- resp.Result
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
var req request
|
||||
json.Unmarshal(line, &req)
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1,
|
||||
"sdk_version": "test",
|
||||
"plugin_name": "cb",
|
||||
"pid": os.Getpid(),
|
||||
}})
|
||||
case "plugin.start":
|
||||
// 在独立 goroutine 里回调,避免阻塞读循环
|
||||
go func(id uint64) {
|
||||
callKernel("tool.register", map[string]interface{}{"name": "cb_tool"})
|
||||
callKernel("settings.get", map[string]interface{}{"key": "some_key"})
|
||||
send(response{ID: id})
|
||||
}(req.ID)
|
||||
case "plugin.stop":
|
||||
send(response{ID: req.ID})
|
||||
out.Flush()
|
||||
os.Exit(0)
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
54
internal/plugin/proc/testdata/crashplugin.go
vendored
Normal file
54
internal/plugin/proc/testdata/crashplugin.go
vendored
Normal file
@ -0,0 +1,54 @@
|
||||
//go:build ignore
|
||||
|
||||
// crashplugin 在收到 boom 工具调用时 panic,用于验证崩溃隔离:
|
||||
// 子进程死亡不应带崩 homed,且内核须能感知退出(供 recordCrash 使用)。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"os"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
out := bufio.NewWriter(os.Stdout)
|
||||
send := func(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
}
|
||||
|
||||
for in.Scan() {
|
||||
var req request
|
||||
if err := json.Unmarshal(in.Bytes(), &req); err != nil {
|
||||
continue
|
||||
}
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1, "sdk_version": "test", "plugin_name": "crash", "pid": os.Getpid(),
|
||||
}})
|
||||
case "tool.invoke":
|
||||
// 模拟插件 bug:直接 panic,进程带非零码退出
|
||||
panic("插件内部 panic:用于验证崩溃隔离")
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
76
internal/plugin/proc/testdata/echoplugin.go
vendored
Normal file
76
internal/plugin/proc/testdata/echoplugin.go
vendored
Normal file
@ -0,0 +1,76 @@
|
||||
//go:build ignore
|
||||
|
||||
// echoplugin 是测试用的最简子进程插件:实现握手 + 回显工具。
|
||||
// 不 import 公开 SDK——只验证 proc 包的 RPC 机制本身。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
in.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
out := bufio.NewWriter(os.Stdout)
|
||||
|
||||
send := func(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
}
|
||||
|
||||
for in.Scan() {
|
||||
var req request
|
||||
if err := json.Unmarshal(in.Bytes(), &req); err != nil {
|
||||
continue
|
||||
}
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1,
|
||||
"sdk_version": "test",
|
||||
"plugin_name": "echo",
|
||||
"pid": os.Getpid(),
|
||||
}})
|
||||
case "plugin.stop":
|
||||
send(response{ID: req.ID})
|
||||
out.Flush()
|
||||
os.Exit(0)
|
||||
case "tool.invoke":
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Args map[string]interface{} `json:"args"`
|
||||
}
|
||||
json.Unmarshal(req.Params, &p)
|
||||
switch p.Name {
|
||||
case "fail_tool":
|
||||
send(response{ID: req.ID, Error: "故意失败:用于验证错误上报"})
|
||||
default:
|
||||
text, _ := p.Args["text"].(string)
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{"result": text}})
|
||||
}
|
||||
case "event.deliver":
|
||||
// 通知:不回应答
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID, Error: fmt.Sprintf("未实现 %s", req.Method)})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
56
internal/plugin/proc/testdata/hangplugin.go
vendored
Normal file
56
internal/plugin/proc/testdata/hangplugin.go
vendored
Normal file
@ -0,0 +1,56 @@
|
||||
//go:build ignore
|
||||
|
||||
// hangplugin 收到工具调用后永久阻塞,用于验证:
|
||||
// 1. 调用方能凭 context 超时返回(不被拖死)
|
||||
// 2. Kill 能真正回收资源(对比 cgo 超时后 OS 线程永久泄漏,§9.3)
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
out := bufio.NewWriter(os.Stdout)
|
||||
send := func(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
}
|
||||
|
||||
for in.Scan() {
|
||||
var req request
|
||||
if err := json.Unmarshal(in.Bytes(), &req); err != nil {
|
||||
continue
|
||||
}
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1, "sdk_version": "test", "plugin_name": "hang", "pid": os.Getpid(),
|
||||
}})
|
||||
case "tool.invoke":
|
||||
// 永久卡住,永不回应答
|
||||
time.Sleep(10 * time.Minute)
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user