docs(plugin-arch): 归档插件架构迁移评估 + plan 第11节整改计划

- docs/zh/架构迁移评估.md: C ABI→子进程+共享内存完整迁移论证(1621行)
- docs/zh/experiments/: 18项可复跑可行性实验(架构评估的所有数字来源)
- plan.md §11: 11.1~11.9 插件架构缺陷修复清单(唯一权威编号)
- main 保持干净,本批次为 update 特性分支的整改起点
This commit is contained in:
dev
2026-08-31 11:45:52 +08:00
parent 48b5c2401c
commit 304cad0648
34 changed files with 3552 additions and 0 deletions

View File

@ -0,0 +1,72 @@
//go:build ignore
package main
/*
#cgo LDFLAGS: -ldl
#include <dlfcn.h>
#include <stdlib.h>
typedef const char* (*verfn)(void);
static const char* call_ver(void* f){ return ((verfn)f)(); }
*/
import "C"
import (
"fmt"
"os"
"unsafe"
)
func main() {
// Go 用 dlopen 加载纯 C shimshim 本身常驻,无所谓)
sp := C.CString("./shim.so")
shim := C.dlopen(sp, C.RTLD_NOW|C.RTLD_LOCAL)
C.free(unsafe.Pointer(sp))
if shim == nil {
fmt.Println("shim 加载失败:", C.GoString(C.dlerror()))
os.Exit(1)
}
openName := C.CString("shim_open")
closeName := C.CString("shim_close")
symName := C.CString("shim_sym")
shimOpen := C.dlsym(shim, openName)
shimClose := C.dlsym(shim, closeName)
shimSym := C.dlsym(shim, symName)
C.free(unsafe.Pointer(openName))
C.free(unsafe.Pointer(closeName))
C.free(unsafe.Pointer(symName))
fmt.Printf("shim 就绪: open=%p close=%p sym=%p\n\n", shimOpen, shimClose, shimSym)
// 直接用 dlopen/dlsym 调 shim 的三个函数(避免再写一层 C 包装)
load := func(path string) unsafe.Pointer {
cp := C.CString(path)
defer C.free(unsafe.Pointer(cp))
return C.dlopen(cp, C.RTLD_NOW|C.RTLD_LOCAL)
}
ver := func(h unsafe.Pointer) string {
n := C.CString("probe_version")
defer C.free(unsafe.Pointer(n))
f := C.dlsym(h, n)
if f == nil { return "<no sym>" }
return C.GoString(C.call_ver(f))
}
fmt.Println("--- 场景: Go(带 NODELETE runtime) 加载/卸载纯 C 的第三层 so ---")
h1 := load("./probe.so")
fmt.Printf("1) dlopen probe.so handle=%p version=%s\n", h1, ver(h1))
rc := C.dlclose(h1)
fmt.Printf("2) dlclose rc=%d\n", int(rc))
// 换内容V1 -> V2同路径
in, _ := os.ReadFile("probe_v2.so")
os.WriteFile("probe.so", in, 0755)
fmt.Println("3) 磁盘 probe.so 内容替换为 V2同路径")
h2 := load("./probe.so")
fmt.Printf("4) 再 dlopen 同路径 handle=%p version=%s\n", h2, ver(h2))
if h1 == h2 {
fmt.Println(" => 句柄相同:未卸载,仍是旧代码")
} else {
fmt.Println(" => 句柄不同:真正卸载并重新装载了新代码 ✅")
}
}

View File

@ -0,0 +1,49 @@
//go:build ignore
package main
/*
#cgo LDFLAGS: -ldl
#include <dlfcn.h>
#include <stdlib.h>
typedef void* (*openfn)(const char*);
typedef int (*closefn)(void*);
static void* c_open(void* f, const char* p){ return ((openfn)f)(p); }
static int c_close(void* f, void* h){ return ((closefn)f)(h); }
*/
import "C"
import (
"fmt"
"os"
"strings"
"unsafe"
)
func cnt(s string) int {
b, _ := os.ReadFile("/proc/self/maps")
n := 0
for _, l := range strings.Split(string(b), "\n") { if strings.Contains(l, s) { n++ } }
return n
}
func main() {
sp := C.CString("./shim.so")
shim := C.dlopen(sp, C.RTLD_NOW|C.RTLD_LOCAL)
C.free(unsafe.Pointer(sp))
no := C.CString("shim_open"); nc := C.CString("shim_close")
fo := C.dlsym(shim, no); fc := C.dlsym(shim, nc)
C.free(unsafe.Pointer(no)); C.free(unsafe.Pointer(nc))
// 经【纯 C shim】去 dlopen/dlclose Go c-shared 插件
qp := C.CString("/home/newqqagent/plugins/qq/plugin.so")
h := C.c_open(fo, qp)
C.free(unsafe.Pointer(qp))
fmt.Printf("经 C shim dlopen Go 插件 handle=%p 映射段=%d\n", h, cnt("qq/plugin.so"))
rc := C.c_close(fc, h)
fmt.Printf("经 C shim dlclose rc=%d 映射段=%d\n", int(rc), cnt("qq/plugin.so"))
if cnt("qq/plugin.so") > 0 {
fmt.Println("\n❌ 仍未卸载 —— NODELETE 属于目标 .so 本身,与谁调 dlopen 无关")
} else {
fmt.Println("\n✅ 卸载成功")
}
}

View File

@ -0,0 +1,58 @@
//go:build ignore
package main
/*
#cgo LDFLAGS: -ldl
#include <dlfcn.h>
#include <stdlib.h>
typedef char* (*verfn)(void);
static char* call_ver(void* f){ return ((verfn)f)(); }
*/
import "C"
import (
"fmt"
"os"
"strings"
"unsafe"
)
func threads() int {
e, _ := os.ReadDir("/proc/self/task")
return len(e)
}
func rss() int {
b, _ := os.ReadFile("/proc/self/status")
for _, l := range strings.Split(string(b), "\n") {
if strings.HasPrefix(l, "VmRSS:") {
var k int
fmt.Sscanf(l, "VmRSS: %d kB", &k)
return k
}
}
return 0
}
func main() {
base, baseT := rss(), threads()
fmt.Printf("基线: RSS=%dKB threads=%d\n\n", base, baseT)
src, _ := os.ReadFile("glv1.so")
os.MkdirAll("stress", 0755)
var hs []unsafe.Pointer
for i := 1; i <= 30; i++ {
p := fmt.Sprintf("stress/%010d-qq.so", 1700000000+i)
os.WriteFile(p, src, 0755)
cp := C.CString("./" + p)
h := C.dlopen(cp, C.RTLD_NOW|C.RTLD_LOCAL)
C.free(unsafe.Pointer(cp))
if h == nil { fmt.Printf("第 %d 次失败\n", i); break }
hs = append(hs, h)
C.dlclose(h) // 模拟每次都尝试卸载no-op
if i%10 == 0 {
fmt.Printf("第 %2d 次重载: RSS=%dKB (+%dKB) threads=%d (+%d)\n",
i, rss(), rss()-base, threads(), threads()-baseT)
}
}
fmt.Printf("\n30 次重载后: RSS 增长 %dKB, 线程增长 %d\n", rss()-base, threads()-baseT)
fmt.Printf("每次重载均摊: RSS +%.1fKB, 线程 +%.2f\n",
float64(rss()-base)/30, float64(threads()-baseT)/30)
}

View File

@ -0,0 +1,2 @@
#include <stdio.h>
const char* probe_version(void){ return "V1"; }

View File

@ -0,0 +1,2 @@
#include <stdio.h>
const char* probe_version(void){ return "V2"; }

View File

@ -0,0 +1,9 @@
#include <dlfcn.h>
#include <stdio.h>
void* shim_open(const char* p){
void* h = dlopen(p, RTLD_NOW|RTLD_LOCAL);
if(!h) printf(" [shim] open FAIL: %s\n", dlerror());
return h;
}
int shim_close(void* h){ return dlclose(h); }
void* shim_sym(void* h, const char* n){ return dlsym(h, n); }

View File

@ -0,0 +1,49 @@
//go:build ignore
package main
import (
"encoding/base64"
"encoding/json"
"fmt"
"time"
"golang.org/x/sys/unix"
)
func main() {
fmt.Println("=== 实验 10多媒体 payload —— 共享内存零拷贝 vs JSON base64 ===")
sizes := []int{100 * 1024, 1024 * 1024, 5 * 1024 * 1024}
for _, sz := range sizes {
img := make([]byte, sz)
for i := range img { img[i] = byte(i % 251) }
// A. JSON + base64当前 ContentBlock 的做法)
t0 := time.Now()
b64 := base64.StdEncoding.EncodeToString(img)
blob, _ := json.Marshal(map[string]string{"type": "image_url", "url": "data:image/png;base64," + b64})
var back map[string]string
json.Unmarshal(blob, &back)
dec, _ := base64.StdEncoding.DecodeString(back["url"][22:])
jsonDur := time.Since(t0)
// B. 共享内存 arena写入 + 偏移解引用,零拷贝读)
mfd, _ := unix.MemfdCreate("arena", 0)
unix.Ftruncate(mfd, int64(sz+4096))
data, _ := unix.Mmap(mfd, 0, sz+4096, unix.PROT_READ|unix.PROT_WRITE, unix.MAP_SHARED)
t0 = time.Now()
copy(data[4096:], img) // 写 arena
view := data[4096 : 4096+sz] // 偏移解引用 = 零拷贝切片
_ = view[sz-1]
shmDur := time.Since(t0)
unix.Munmap(data)
unix.Close(mfd)
fmt.Printf("\n%s payload:\n", map[int]string{100*1024:"100KB", 1024*1024:"1MB", 5*1024*1024:"5MB"}[sz])
fmt.Printf(" A JSON+base64: %8v 传输体积 %d B (+%.0f%%) 解出 %d B %s\n",
jsonDur, len(blob), float64(len(blob)-sz)/float64(sz)*100, len(dec),
map[bool]string{true:"✓",false:"✗"}[len(dec)==sz])
fmt.Printf(" B 共享内存: %8v 传输体积 8 B (描述符) 零拷贝视图 %d B\n", shmDur, len(view))
fmt.Printf(" → 加速 %.0fx, 体积节省 %.0f%%\n",
float64(jsonDur)/float64(shmDur), float64(len(blob)-8)/float64(len(blob))*100)
}
}

View File

@ -0,0 +1,42 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"fmt"
"os/exec"
"sort"
"time"
)
type Req struct{ ID int `json:"id"`; Method string `json:"method"`; Args json.RawMessage `json:"args"` }
type Res struct{ ID int `json:"id"`; Result string `json:"result"` }
func main() {
fmt.Println("=== 实验 11工具调用 RPC 端到端延迟(实测 payload 中位 93B===")
cmd := exec.Command("./plug11")
sin, _ := cmd.StdinPipe(); sout, _ := cmd.StdoutPipe()
cmd.Start()
enc := json.NewEncoder(bufio.NewWriter(sin))
w := bufio.NewWriter(sin); enc = json.NewEncoder(w)
dec := json.NewDecoder(bufio.NewReader(sout))
args := json.RawMessage(`{"city":"hangzhou","days":3,"unit":"celsius","detail":true}`)
const N = 10000
lat := make([]time.Duration, 0, N)
for i := 0; i < N; i++ {
t0 := time.Now()
enc.Encode(Req{ID: i, Method: "weather_query", Args: args}); w.Flush()
var r Res
if err := dec.Decode(&r); err != nil { break }
lat = append(lat, time.Since(t0))
}
sin.Close(); cmd.Wait()
sort.Slice(lat, func(a,b int) bool { return lat[a] < lat[b] })
p := func(q float64) time.Duration { return lat[int(float64(len(lat))*q)] }
fmt.Printf("样本 %d 次\n", len(lat))
fmt.Printf(" p50 = %v\n p90 = %v\n p99 = %v\n max = %v\n", p(0.5), p(0.9), p(0.99), lat[len(lat)-1])
fmt.Printf("\n对照 LLM 单轮往返 2-8 秒 → RPC 占比 ≈ %.5f%%\n",
float64(p(0.5))/float64(3*time.Second)*100)
}

View File

@ -0,0 +1,12 @@
//go:build ignore
package main
import ("bufio";"encoding/json";"os")
type Req struct{ ID int `json:"id"`; Method string `json:"method"`; Args json.RawMessage `json:"args"` }
type Res struct{ ID int `json:"id"`; Result string `json:"result"` }
func main(){
dec:=json.NewDecoder(bufio.NewReader(os.Stdin))
w:=bufio.NewWriter(os.Stdout); enc:=json.NewEncoder(w)
for { var q Req
if err:=dec.Decode(&q); err!=nil {return}
enc.Encode(Res{ID:q.ID, Result:`{"ok":true,"data":"` + string(q.Args) + `"}`}); w.Flush() }
}

View File

@ -0,0 +1,58 @@
//go:build ignore
package main
import (
"fmt"
"os"
"runtime"
"sync"
"sync/atomic"
"time"
"golang.org/x/sys/unix"
)
func threads() int { e, _ := os.ReadDir("/proc/self/task"); return len(e) }
func main() {
fmt.Println("=== 实验 1eventfd 是否走 Go netpoller只 park goroutine 不占 OS 线程)===")
base := threads()
fmt.Printf("基线线程数: %d (GOMAXPROCS=%d)\n\n", base, runtime.GOMAXPROCS(0))
const N = 200 // 模拟 200 个订阅者等待
var wg sync.WaitGroup
var woke int64
files := make([]*os.File, N)
for i := 0; i < N; i++ {
efd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
if err != nil { fmt.Println("eventfd 失败:", err); return }
f := os.NewFile(uintptr(efd), fmt.Sprintf("evt%d", i))
files[i] = f
wg.Add(1)
go func(f *os.File) {
defer wg.Done()
buf := make([]byte, 8)
// 阻塞读:若走 netpoller 只 park goroutine
if _, err := f.Read(buf); err == nil {
atomic.AddInt64(&woke, 1)
}
}(f)
}
time.Sleep(500 * time.Millisecond) // 让所有 goroutine 进入等待
waiting := threads()
fmt.Printf("%d 个 goroutine 阻塞在 eventfd.Read 后:\n", N)
fmt.Printf(" 线程数 = %d (增长 %d)\n", waiting, waiting-base)
if waiting-base < 20 {
fmt.Println(" ✅ 走 netpoller线程未随等待者数量增长")
} else {
fmt.Printf(" ❌ 退化为阻塞 syscall每个等待者占一个 OS 线程\n")
}
// 全部唤醒
one := []byte{1,0,0,0,0,0,0,0}
for _, f := range files { f.Write(one) }
wg.Wait()
fmt.Printf("\n唤醒数 = %d/%d 唤醒后线程数 = %d\n", woke, N, threads())
}

View File

@ -0,0 +1,41 @@
//go:build ignore
package main
import (
"encoding/binary"
"fmt"
"os"
"unsafe"
"golang.org/x/sys/unix"
)
// 子进程fd 3 = eventfd(通知), fd 4 = shm 文件
func main() {
efd := os.NewFile(3, "evt")
shmf := os.NewFile(4, "shm")
data, err := unix.Mmap(int(shmf.Fd()), 0, 4096, unix.PROT_READ|unix.PROT_WRITE, unix.MAP_SHARED)
if err != nil { fmt.Println("CHILD mmap 失败:", err); os.Exit(1) }
fmt.Printf("CHILD: mmap 基址 = %p\n", unsafe.Pointer(&data[0]))
buf := make([]byte, 8)
if _, err := efd.Read(buf); err != nil {
fmt.Println("CHILD read err:", err); os.Exit(1)
}
n := binary.LittleEndian.Uint64(buf)
fmt.Printf("CHILD: 被 eventfd 唤醒, 计数=%d\n", n)
// 按偏移读:头部 16 字节 = {off uint32, len uint32, seq uint64}
off := binary.LittleEndian.Uint32(data[0:4])
ln := binary.LittleEndian.Uint32(data[4:8])
seq := binary.LittleEndian.Uint64(data[8:16])
payload := string(data[off : off+ln])
fmt.Printf("CHILD: 偏移解引用 off=%d len=%d seq=%d → %q\n", off, ln, seq, payload)
// 子进程回写(验证双向可见)
copy(data[2048:], []byte("CHILD-ACK"))
binary.LittleEndian.PutUint32(data[16:20], 2048)
binary.LittleEndian.PutUint32(data[20:24], uint32(len("CHILD-ACK")))
fmt.Println("CHILD: 已回写 ACK")
}

View File

@ -0,0 +1,60 @@
//go:build ignore
package main
import (
"encoding/binary"
"fmt"
"os"
"os/exec"
"time"
"unsafe"
"golang.org/x/sys/unix"
)
func main() {
fmt.Println("=== 实验 2跨进程 eventfd 通知 + 共享内存偏移解引用 ===")
// eventfd 不带 CLOEXEC需要被子进程继承
efd, err := unix.Eventfd(0, unix.EFD_NONBLOCK)
if err != nil { panic(err) }
evtFile := os.NewFile(uintptr(efd), "evt")
// shm: 用 memfd匿名无需 /dev/shm 清理)
mfd, err := unix.MemfdCreate("stagectx", 0)
if err != nil { panic(err) }
if err := unix.Ftruncate(mfd, 4096); err != nil { panic(err) }
shmFile := os.NewFile(uintptr(mfd), "shm")
data, err := unix.Mmap(mfd, 0, 4096, unix.PROT_READ|unix.PROT_WRITE, unix.MAP_SHARED)
if err != nil { panic(err) }
fmt.Printf("PARENT: mmap 基址 = %p\n", unsafe.Pointer(&data[0]))
// 写 payload 到 arena(偏移 1024),头部记描述符
msg := "hello-from-parent-via-offset"
copy(data[1024:], []byte(msg))
binary.LittleEndian.PutUint32(data[0:4], 1024)
binary.LittleEndian.PutUint32(data[4:8], uint32(len(msg)))
binary.LittleEndian.PutUint64(data[8:16], 42)
fmt.Printf("PARENT: 数据已落地 arena@1024, 描述符 {off:1024, len:%d, seq:42}\n", len(msg))
cmd := exec.Command("go", "run", "exp2_child.go")
cmd.ExtraFiles = []*os.File{evtFile, shmFile} // → 子进程 fd 3, 4
cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
if err := cmd.Start(); err != nil { panic(err) }
time.Sleep(3 * time.Second) // 等 go run 编译+启动
fmt.Println("PARENT: 数据到位后 post eventfd不等待消费者")
t0 := time.Now()
evtFile.Write([]byte{1,0,0,0,0,0,0,0})
fmt.Printf("PARENT: post 耗时 %v ← post-and-forget\n", time.Since(t0))
cmd.Wait()
// 读子进程回写
off := binary.LittleEndian.Uint32(data[16:20])
ln := binary.LittleEndian.Uint32(data[20:24])
if ln > 0 {
fmt.Printf("PARENT: 读到子进程回写 → %q ✅ 双向可见\n", string(data[off:off+ln]))
}
}

View File

@ -0,0 +1,31 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"fmt"
"os"
"time"
)
type req struct{ ID int `json:"id"`; Method string `json:"method"` }
type resp struct{ ID int `json:"id"`; OK bool `json:"ok"` }
func main() {
in := bufio.NewReader(os.Stdin)
out := bufio.NewWriter(os.Stdout)
enc, dec := json.NewEncoder(out), json.NewDecoder(in)
const N = 20000
t0 := time.Now()
for i := 0; i < N; i++ {
enc.Encode(req{ID: i, Method: "stage.lock"})
out.Flush()
var r resp
if err := dec.Decode(&r); err != nil { fmt.Fprintln(os.Stderr, "dec:", err); return }
}
d := time.Since(t0)
fmt.Fprintf(os.Stderr, "CHILD: %d 次 lock RPC 往返 用时 %v, 均摊 %.2f µs/次\n",
N, d, float64(d.Microseconds())/float64(N))
}

View File

@ -0,0 +1,37 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"fmt"
"os"
"os/exec"
"sync"
)
type req struct{ ID int `json:"id"`; Method string `json:"method"` }
type resp struct{ ID int `json:"id"`; OK bool `json:"ok"` }
func main() {
fmt.Println("=== 实验 3锁仲裁 RPC 往返成本stdio JSON-RPC===")
cmd := exec.Command("go", "run", "exp3_child.go")
stdin, _ := cmd.StdinPipe()
stdout, _ := cmd.StdoutPipe()
cmd.Stderr = os.Stderr
cmd.Start()
var mu sync.Mutex // 内核侧真实的锁仲裁
dec := json.NewDecoder(bufio.NewReader(stdout))
w := bufio.NewWriter(stdin)
enc := json.NewEncoder(w)
for {
var q req
if err := dec.Decode(&q); err != nil { break }
mu.Lock() // 真实加锁
mu.Unlock() // 立即释放(模拟仲裁开销)
enc.Encode(resp{ID: q.ID, OK: true})
w.Flush()
}
cmd.Wait()
}

View File

@ -0,0 +1,71 @@
//go:build ignore
package main
import (
"fmt"
"os"
"sync/atomic"
"time"
"golang.org/x/sys/unix"
)
type ring struct {
writeSeq atomic.Uint64
cap uint64
slots []uint64
}
func main() {
fmt.Println("=== 实验 4事件环 post-and-forget vs 同步 Publish慢消费者场景===")
const tokens = 5000
// --- A. 现状:同步 Publish消费者慢 ---
slowHandler := func() { time.Sleep(20 * time.Microsecond) }
t0 := time.Now()
for i := 0; i < tokens; i++ { slowHandler() }
syncDur := time.Since(t0)
fmt.Printf("A 同步 Publish (慢消费者 20µs): %d token 耗时 %v → 均摊 %.1f µs/token\n",
tokens, syncDur, float64(syncDur.Microseconds())/tokens)
// --- B. 新方案:写环 + eventfd post不等消费者 ---
r := &ring{cap: 1024, slots: make([]uint64, 1024)}
efd, _ := unix.Eventfd(0, unix.EFD_NONBLOCK)
f := os.NewFile(uintptr(efd), "e")
var dropped atomic.Uint64
// 慢消费者 goroutine
done := make(chan struct{})
go func() {
buf := make([]byte, 8)
var readSeq uint64
for {
if _, err := f.Read(buf); err != nil { return }
w := r.writeSeq.Load()
if w-readSeq > r.cap {
dropped.Add(w - readSeq - r.cap)
readSeq = w - r.cap
}
for readSeq < w { readSeq++ }
time.Sleep(20 * time.Microsecond) // 慢
select { case <-done: return; default: }
}
}()
t0 = time.Now()
one := []byte{1,0,0,0,0,0,0,0}
for i := 0; i < tokens; i++ {
s := r.writeSeq.Add(1)
r.slots[s%r.cap] = s // 写数据
f.Write(one) // post不等
}
asyncDur := time.Since(t0)
close(done)
fmt.Printf("B 环+eventfd post: %d token 耗时 %v → 均摊 %.2f µs/token\n",
tokens, asyncDur, float64(asyncDur.Microseconds())/tokens)
fmt.Printf("\n加速比 %.1fx 丢弃事件 %d消费者跟不上已计数\n",
float64(syncDur)/float64(asyncDur), dropped.Load())
if asyncDur < syncDur/5 {
fmt.Println("✅ post-and-forget 使流式发布与消费者速度解耦")
}
}

View File

@ -0,0 +1,55 @@
//go:build ignore
package main
import (
"fmt"
"os"
"os/exec"
"strconv"
"strings"
"time"
)
func pssKB(pid int) int {
b, err := os.ReadFile(fmt.Sprintf("/proc/%d/smaps_rollup", pid))
if err != nil { return 0 }
for _, l := range strings.Split(string(b), "\n") {
if strings.HasPrefix(l, "Pss:") {
f := strings.Fields(l)
n, _ := strconv.Atoi(f[1]); return n
}
}
return 0
}
func threads(pid int) int {
e, _ := os.ReadDir(fmt.Sprintf("/proc/%d/task", pid)); return len(e)
}
func main() {
fmt.Println("=== 实验 517 个 Go 子进程插件的真实常驻开销PSS 计入共享页去重)===")
var cmds []*exec.Cmd
for i := 0; i < 17; i++ {
c := exec.Command("./plugbin")
c.Stdin, _ = os.Open(os.DevNull)
if err := c.Start(); err != nil { fmt.Println("start:", err); return }
cmds = append(cmds, c)
}
time.Sleep(1500 * time.Millisecond)
totalPss, totalThreads := 0, 0
for _, c := range cmds {
totalPss += pssKB(c.Process.Pid)
totalThreads += threads(c.Process.Pid)
}
fmt.Printf("17 进程合计: PSS = %.1f MB, 线程 = %d\n", float64(totalPss)/1024, totalThreads)
fmt.Printf("单进程均摊: PSS = %.2f MB, 线程 = %.1f\n",
float64(totalPss)/1024/17, float64(totalThreads)/17)
fmt.Printf("\n对照 homed 当前(单进程装 17 个 .so:\n")
// 找 homed
out, _ := exec.Command("pgrep", "-x", "homed").Output()
if p := strings.TrimSpace(string(out)); p != "" {
pid, _ := strconv.Atoi(strings.Fields(p)[0])
fmt.Printf(" homed PSS = %.1f MB, 线程 = %d\n", float64(pssKB(pid))/1024, threads(pid))
}
for _, c := range cmds { c.Process.Kill(); c.Wait() }
}

View File

@ -0,0 +1,23 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"os"
)
// 模拟一个最小插件stdio JSON-RPC loop + 一个 goroutine
func main() {
go func() { select {} }()
in := bufio.NewReader(os.Stdin)
dec := json.NewDecoder(in)
out := bufio.NewWriter(os.Stdout)
enc := json.NewEncoder(out)
for {
var m map[string]interface{}
if err := dec.Decode(&m); err != nil { return }
enc.Encode(map[string]interface{}{"ok": true})
out.Flush()
}
}

View File

@ -0,0 +1,68 @@
//go:build ignore
package main
import (
"fmt"
"os"
"os/exec"
"strconv"
"strings"
"time"
)
func pssKB(pid int) int {
b, err := os.ReadFile(fmt.Sprintf("/proc/%d/smaps_rollup", pid))
if err != nil { return -1 }
for _, l := range strings.Split(string(b), "\n") {
if strings.HasPrefix(l, "Pss:") { f := strings.Fields(l); n,_ := strconv.Atoi(f[1]); return n }
}
return -1
}
func rssKB(pid int) int {
b, err := os.ReadFile(fmt.Sprintf("/proc/%d/status", pid))
if err != nil { return -1 }
for _, l := range strings.Split(string(b), "\n") {
if strings.HasPrefix(l, "VmRSS:") { f := strings.Fields(l); n,_ := strconv.Atoi(f[1]); return n }
}
return -1
}
func threads(pid int) int { e,_ := os.ReadDir(fmt.Sprintf("/proc/%d/task", pid)); return len(e) }
func main() {
fmt.Println("=== 实验 5b17 个 Go 子进程常驻开销(保持 stdin 管道存活)===")
var cmds []*exec.Cmd
var pipes []interface{ Close() error }
for i := 0; i < 17; i++ {
c := exec.Command("./plugbin")
w, _ := c.StdinPipe() // 保持打开 → 不 EOF
pipes = append(pipes, w)
c.Stdout = nil
if err := c.Start(); err != nil { fmt.Println(err); return }
cmds = append(cmds, c)
}
time.Sleep(2 * time.Second)
tp, tr, tt, alive := 0, 0, 0, 0
for _, c := range cmds {
pid := c.Process.Pid
if _, err := os.Stat(fmt.Sprintf("/proc/%d", pid)); err != nil { continue }
alive++
if v := pssKB(pid); v > 0 { tp += v }
if v := rssKB(pid); v > 0 { tr += v }
tt += threads(pid)
}
fmt.Printf("存活进程 %d/17\n", alive)
fmt.Printf("合计: PSS=%.1f MB RSS=%.1f MB 线程=%d\n",
float64(tp)/1024, float64(tr)/1024, tt)
if alive > 0 {
fmt.Printf("均摊: PSS=%.2f MB RSS=%.2f MB 线程=%.1f\n",
float64(tp)/1024/float64(alive), float64(tr)/1024/float64(alive), float64(tt)/float64(alive))
}
out, _ := exec.Command("pgrep", "-x", "homed").Output()
if p := strings.TrimSpace(string(out)); p != "" {
pid, _ := strconv.Atoi(strings.Fields(p)[0])
fmt.Printf("\n对照 homed单进程 + 17 个 .so: RSS=%.1f MB 线程=%d\n",
float64(rssKB(pid))/1024, threads(pid))
}
for _, c := range cmds { c.Process.Kill(); c.Wait() }
}

View File

@ -0,0 +1,49 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/exec"
"time"
)
func main() {
fmt.Println("=== 实验 6子进程崩溃隔离 + 退出码/EOF 作为 recordCrash 信号 ===")
cmd := exec.Command("./crashbin")
sin, _ := cmd.StdinPipe()
sout, _ := cmd.StdoutPipe()
cmd.Stderr = nil // 丢弃 panic 栈
cmd.Start()
fmt.Printf("插件进程 pid=%d 已启动\n", cmd.Process.Pid)
enc := json.NewEncoder(sin)
dec := json.NewDecoder(bufio.NewReader(sout))
// 正常调用
enc.Encode(map[string]string{"method": "ping"})
var r map[string]interface{}
if err := dec.Decode(&r); err == nil { fmt.Println("正常调用 → ", r) }
// 触发崩溃
fmt.Println("\n发送 boom插件内 panic...")
t0 := time.Now()
enc.Encode(map[string]string{"method": "boom"})
err := dec.Decode(&r)
detected := "未检测到"
if errors.Is(err, io.EOF) || err == io.ErrUnexpectedEOF { detected = "EOF" } else if err != nil { detected = fmt.Sprintf("%v", err) }
fmt.Printf("调用侧感知: %s (耗时 %v)\n", detected, time.Since(t0))
werr := cmd.Wait()
var ec int = -1
if ee, ok := werr.(*exec.ExitError); ok { ec = ee.ExitCode() }
fmt.Printf("进程退出码 = %d panic → 2可直接喂 recordCrash\n", ec)
fmt.Printf("\n宿主进程仍存活: pid=%d ✅ 崩溃已隔离\n", os.Getpid())
fmt.Println("→ 对照:当前 .so 模型下bridge 兜不住的 panic 会带崩整个 homed")
}

View File

@ -0,0 +1,23 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"os"
)
func main() {
dec := json.NewDecoder(bufio.NewReader(os.Stdin))
out := bufio.NewWriter(os.Stdout)
enc := json.NewEncoder(out)
for {
var m map[string]interface{}
if err := dec.Decode(&m); err != nil { return }
if m["method"] == "boom" {
panic("插件故意崩溃") // 真 panic
}
enc.Encode(map[string]interface{}{"ok": true})
out.Flush()
}
}

View File

@ -0,0 +1,63 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"fmt"
"os"
"os/exec"
"time"
)
func spawnAndAsk(bin string) string {
cmd := exec.Command(bin)
sin, _ := cmd.StdinPipe()
sout, _ := cmd.StdoutPipe()
cmd.Start()
enc := json.NewEncoder(sin)
dec := json.NewDecoder(bufio.NewReader(sout))
enc.Encode(map[string]string{"method": "version"})
var r map[string]interface{}
dec.Decode(&r)
sin.Close()
cmd.Process.Kill()
cmd.Wait()
if v, ok := r["version"].(string); ok { return v }
return "?"
}
func build(ver, out string) {
src := fmt.Sprintf(`package main
import ("bufio";"encoding/json";"os")
func main(){
dec:=json.NewDecoder(bufio.NewReader(os.Stdin))
w:=bufio.NewWriter(os.Stdout); enc:=json.NewEncoder(w)
for { var m map[string]interface{}
if err:=dec.Decode(&m); err!=nil {return}
enc.Encode(map[string]string{"version":%q}); w.Flush() }
}`, ver)
os.MkdirAll("v", 0755)
os.WriteFile("v/main.go", []byte(src), 0644)
os.WriteFile("v/go.mod", []byte("module v\ngo 1.21\n"), 0644)
c := exec.Command("go", "build", "-o", "../"+out, ".")
c.Dir = "v"
if b, err := c.CombinedOutput(); err != nil { fmt.Println("build err:", string(b)) }
}
func main() {
fmt.Println("=== 实验 7子进程模型下的热重载迁移的原始目标===")
build("v1.0.0", "hotbin")
fmt.Printf("1) 首次启动插件 → version = %s\n", spawnAndAsk("./hotbin"))
fmt.Println("2) 替换二进制为 v2.0.0(同路径,无需版本化 hash 目录)")
build("v2.0.0", "hotbin")
time.Sleep(200 * time.Millisecond)
v := spawnAndAsk("./hotbin")
fmt.Printf("3) 重启插件进程 → version = %s\n", v)
if v == "v2.0.0" {
fmt.Println("\n✅ 同路径替换即生效:无 NODELETE、无版本化路径、无线程泄漏")
fmt.Println(" 对照 .so 模型:同路径 dlopen 复用旧映像,永远拿不到 v2")
}
}

View File

@ -0,0 +1,84 @@
//go:build ignore
package main
import (
"bufio"
"encoding/binary"
"encoding/json"
"fmt"
"os"
"os/exec"
"strings"
"sync"
"time"
"golang.org/x/sys/unix"
)
func main() {
fmt.Println("=== 实验 8跨进程并发扇出改写同一 StageContext最高风险点 3.4===")
mfd, _ := unix.MemfdCreate("stagectx", 0)
unix.Ftruncate(mfd, 65536)
shmFile := os.NewFile(uintptr(mfd), "shm")
data, _ := unix.Mmap(mfd, 0, 65536, unix.PROT_READ|unix.PROT_WRITE, unix.MAP_SHARED)
// 初始 final_text = "" @1024, arena 游标 = 1024
binary.LittleEndian.PutUint32(data[0:4], 1024)
binary.LittleEndian.PutUint32(data[4:8], 0)
binary.LittleEndian.PutUint32(data[8:12], 1024)
tags := []string{"A", "B", "C", "D", "E"} // 5 个并发插件
var mu sync.Mutex // 内核侧锁仲裁
var wg sync.WaitGroup
var rpcCount int64
var cntMu sync.Mutex
t0 := time.Now()
for _, tag := range tags {
cmd := exec.Command("go", "run", "exp8_worker.go", tag)
cmd.ExtraFiles = []*os.File{shmFile}
sin, _ := cmd.StdinPipe()
sout, _ := cmd.StdoutPipe()
cmd.Stderr = os.Stderr
cmd.Start()
wg.Add(1)
go func() {
defer wg.Done()
dec := json.NewDecoder(bufio.NewReader(sout))
w := bufio.NewWriter(sin)
enc := json.NewEncoder(w)
held := false
for {
var q map[string]string
if err := dec.Decode(&q); err != nil { break }
switch q["method"] {
case "stage.lock": mu.Lock(); held = true
case "stage.unlock": if held { mu.Unlock(); held = false }
}
cntMu.Lock(); rpcCount++; cntMu.Unlock()
enc.Encode(map[string]bool{"ok": true}); w.Flush()
}
if held { mu.Unlock() }
cmd.Wait()
}()
}
wg.Wait()
dur := time.Since(t0)
off := binary.LittleEndian.Uint32(data[0:4])
ln := binary.LittleEndian.Uint32(data[4:8])
final := string(data[off : off+ln])
fmt.Printf("\n--- 结果 ---\n")
fmt.Printf("最终 final_text 长度 = %d\n", len(final))
counts := map[string]int{}
for _, t := range tags { counts[t] = strings.Count(final, t) }
fmt.Printf("各插件写入次数: %v\n", counts)
total := 0
for _, c := range counts { total += c }
fmt.Printf("总字符 = %d, 长度 = %d → %s\n", total, len(final),
map[bool]string{true:"一致 ✅ 无丢失/无撕裂", false:"不一致 ❌"}[total == len(final)])
fmt.Printf("RPC 锁操作 = %d 次, 总耗时 %v\n", rpcCount, dur)
fmt.Printf("\n注写入次数少于 5×300 是 arena 64KB 上限所致append-only 未压实),符合设计\n")
}

View File

@ -0,0 +1,50 @@
//go:build ignore
package main
import (
"bufio"
"encoding/binary"
"encoding/json"
"fmt"
"os"
"strconv"
"golang.org/x/sys/unix"
)
// 模拟插件:拿锁 → 读 final_text → 追加自己的标记 → 写回 → 放锁
// 锁通过 stdio RPC 向内核申请(方案 3.7:锁仲裁回归内核,无 cgo
func main() {
tag := os.Args[1]
shmf := os.NewFile(3, "shm")
data, err := unix.Mmap(int(shmf.Fd()), 0, 65536, unix.PROT_READ|unix.PROT_WRITE, unix.MAP_SHARED)
if err != nil { fmt.Fprintln(os.Stderr, "mmap:", err); os.Exit(1) }
dec := json.NewDecoder(bufio.NewReader(os.Stdin))
w := bufio.NewWriter(os.Stdout)
enc := json.NewEncoder(w)
rpc := func(method string) {
enc.Encode(map[string]string{"method": method}); w.Flush()
var r map[string]interface{}; dec.Decode(&r)
}
const iters = 300
for i := 0; i < iters; i++ {
rpc("stage.lock")
// --- 临界区:偏移解引用读写 final_text ---
off := binary.LittleEndian.Uint32(data[0:4])
ln := binary.LittleEndian.Uint32(data[4:8])
cur := string(data[off : off+ln])
add := tag
newS := cur + add
// append-only arena写到新位置
newOff := binary.LittleEndian.Uint32(data[8:12])
if int(newOff)+len(newS) > 65536 { rpc("stage.unlock"); break }
copy(data[newOff:], []byte(newS))
binary.LittleEndian.PutUint32(data[0:4], newOff)
binary.LittleEndian.PutUint32(data[4:8], uint32(len(newS)))
binary.LittleEndian.PutUint32(data[8:12], newOff+uint32(len(newS)))
rpc("stage.unlock")
}
fmt.Fprintln(os.Stderr, "worker "+tag+" done, iters="+strconv.Itoa(iters))
}

View File

@ -0,0 +1,60 @@
//go:build ignore
package main
import (
"bufio"
"encoding/json"
"fmt"
"os/exec"
"sync"
"time"
)
func run(name, arg string, mu *sync.Mutex, crashed *bool) {
cmd := exec.Command("go", "run", "exp9_worker.go", arg)
sin, _ := cmd.StdinPipe(); sout, _ := cmd.StdoutPipe()
cmd.Stderr = nil
cmd.Start()
dec := json.NewDecoder(bufio.NewReader(sout))
w := bufio.NewWriter(sin); enc := json.NewEncoder(w)
held := false
for {
var q map[string]string
if err := dec.Decode(&q); err != nil { break }
switch q["method"] {
case "stage.lock": mu.Lock(); held = true; fmt.Printf(" [%s] 获得锁\n", name)
case "stage.unlock": if held { mu.Unlock(); held = false; fmt.Printf(" [%s] 释放锁\n", name) }
}
enc.Encode(map[string]bool{"ok":true}); w.Flush()
}
err := cmd.Wait()
// 关键:进程死了,内核侧检测到 EOF/退出 → 强制释放它持有的锁
if held {
mu.Unlock()
*crashed = true
fmt.Printf(" [%s] 进程死亡(%v),内核强制释放其持有的锁 ← 自愈\n", name, err)
}
}
func main() {
fmt.Println("=== 实验 9持锁进程崩溃后的自愈验证无需 robust pthread_mutex===")
var mu sync.Mutex
crashed := false
fmt.Println("\n1) 插件 X 拿锁后 panic:")
run("X", "crash", &mu, &crashed)
fmt.Println("\n2) 插件 Y 随后申请同一把锁:")
done := make(chan bool, 1)
go func() { run("Y", "normal", &mu, new(bool)); done <- true }()
select {
case <-done:
fmt.Println("\n✅ Y 正常获得并释放锁 —— 无死锁")
fmt.Println(" → 内核持有锁的所有权,进程死亡由 Wait()/EOF 检测并强制释放")
fmt.Println(" → 不需要 PTHREAD_PROCESS_SHARED|ROBUST也不需要处理 EOWNERDEAD")
fmt.Println(" → 整个架构可做到零 cgo")
case <-time.After(15 * time.Second):
fmt.Println("\n❌ 死锁Y 拿不到锁(说明需要 robust 语义)")
}
_ = crashed
}

View File

@ -0,0 +1,12 @@
//go:build ignore
package main
import ("bufio";"encoding/json";"os")
func main() {
dec := json.NewDecoder(bufio.NewReader(os.Stdin))
w := bufio.NewWriter(os.Stdout); enc := json.NewEncoder(w)
rpc := func(m string) { enc.Encode(map[string]string{"method":m}); w.Flush(); var r map[string]interface{}; dec.Decode(&r) }
rpc("stage.lock")
if os.Args[1] == "crash" { panic("持锁时崩溃") } // 拿着锁死掉
rpc("stage.unlock")
}

View File

@ -0,0 +1,90 @@
//go:build ignore
package main
import (
"encoding/json"
"fmt"
"strings"
"sync"
)
// 完全复刻内核 loader.go case 2 + templates.go go_invoke_stage 的链路
type StageCtx struct {
mu sync.RWMutex
LLMText string
ToolRes []string
}
func (c *StageCtx) Lock() { c.mu.Lock() }
func (c *StageCtx) Unlock() { c.mu.Unlock() }
func (c *StageCtx) RLock() { c.mu.RLock() }
func (c *StageCtx) RUnlock() { c.mu.RUnlock() }
// === 模拟外部插件(副本模型)===
func externalPlugin(tag string, ctxJSON string) string {
// go_invoke_stage: 新建全新对象
sc := &StageCtx{}
var m map[string]interface{}
json.Unmarshal([]byte(ctxJSON), &m)
if v, ok := m["llm_text"].(string); ok { sc.LLMText = v }
// 插件 handlerctx.Lock() 锁的是这个新对象 → 空转
sc.Lock()
sc.LLMText = sc.LLMText + "[" + tag + "]"
sc.Unlock()
out, _ := json.Marshal(map[string]interface{}{"llm_text": sc.LLMText})
return string(out)
}
// === 模拟内核 case 2 handler ===
func kernelStageHandler(sc *StageCtx, tag string) {
sc.RLock()
snap, _ := json.Marshal(map[string]interface{}{"llm_text": sc.LLMText})
sc.RUnlock()
result := externalPlugin(tag, string(snap))
// applyStageResult
var m map[string]interface{}
json.Unmarshal([]byte(result), &m)
sc.Lock()
if v, ok := m["llm_text"].(string); ok { sc.LLMText = v }
sc.Unlock()
}
// === 内置插件:直接改同一对象 ===
func nativePlugin(sc *StageCtx, tag string) {
sc.Lock()
sc.LLMText = sc.LLMText + "[" + tag + "]"
sc.Unlock()
}
func runCase(name string, fn func(*StageCtx, string), tags []string, rounds int) {
lost := 0
for r := 0; r < rounds; r++ {
sc := &StageCtx{LLMText: "BASE"}
var wg sync.WaitGroup
for _, t := range tags {
wg.Add(1)
go func(t string) { defer wg.Done(); fn(sc, t) }(t)
}
wg.Wait()
// 检查是否所有 tag 都在
for _, t := range tags {
if !strings.Contains(sc.LLMText, "["+t+"]") { lost++; break }
}
}
fmt.Printf(" %-28s %d/%d 轮出现修改丢失 (%.1f%%)\n", name, lost, rounds, float64(lost)/float64(rounds)*100)
}
func main() {
tags := []string{"A", "B", "C", "D", "E"}
fmt.Println("5 个插件并发在 StageBeforeToolcall 追加标记,各 2000 轮:")
fmt.Println()
runCase("内置插件(共享同一对象)", nativePlugin, tags, 2000)
runCase("外部插件(快照-副本-写回)", kernelStageHandler, tags, 2000)
fmt.Println()
fmt.Println("→ 副本模型下 read-modify-write 非原子:快照与写回之间的窗口导致覆盖")
}

View File

@ -0,0 +1,101 @@
//go:build ignore
package main
// 精确复刻现网 AfterToolcall 上 sanitizer(Global,改写) + weather(OwnTools,只读) 的并发
import (
"encoding/json"
"fmt"
"strings"
"sync"
)
type ToolResult struct {
Name string `json:"name"`
Plugin string `json:"plugin"`
Result interface{} `json:"result"`
}
type Ctx struct {
mu sync.RWMutex
ToolRes []ToolResult
}
func (c *Ctx) Lock(){c.mu.Lock()}; func (c *Ctx) Unlock(){c.mu.Unlock()}
func (c *Ctx) RLock(){c.mu.RLock()}; func (c *Ctx) RUnlock(){c.mu.RUnlock()}
func cleanText(s string) string {
// 模拟 sanitizer去掉 ANSI/坏字节
return strings.ReplaceAll(s, "\x1b[31m", "")
}
// 内核 case 2 handler外部插件通用路径
func kernelExternal(sc *Ctx, pluginFn func(*Ctx)) {
// 1. 快照
sc.RLock()
snap, _ := json.Marshal(map[string]interface{}{"tool_results": sc.ToolRes})
sc.RUnlock()
// 2. go_invoke_stage: 插件进程内全新对象
local := &Ctx{}
var m map[string]interface{}
json.Unmarshal(snap, &m)
if v, ok := m["tool_results"]; ok {
b, _ := json.Marshal(v)
json.Unmarshal(b, &local.ToolRes)
}
// 3. 插件 handler 跑在副本上
pluginFn(local)
// 4. stageContextWritable: 无条件回传 tool_results
out := map[string]interface{}{}
if len(local.ToolRes) > 0 { out["tool_results"] = local.ToolRes }
rb, _ := json.Marshal(out)
// 5. applyStageResult 写回内核
var rm map[string]interface{}
json.Unmarshal(rb, &rm)
sc.Lock()
if v, ok := rm["tool_results"]; ok {
b, _ := json.Marshal(v)
var trs []ToolResult
if json.Unmarshal(b, &trs) == nil { sc.ToolRes = trs }
}
sc.Unlock()
}
func sanitizerStage(ctx *Ctx) {
ctx.Lock(); defer ctx.Unlock()
for i, tr := range ctx.ToolRes {
if s, ok := tr.Result.(string); ok {
ctx.ToolRes[i].Result = cleanText(s)
}
}
}
func weatherStage(ctx *Ctx) {
ctx.Lock(); defer ctx.Unlock()
// 只读打印不改own_tools scope 已匹配)
_ = len(ctx.ToolRes)
}
func main() {
const rounds = 3000
dirty := "\x1b[31m晴 25°C"
polluted := 0
for r := 0; r < rounds; r++ {
sc := &Ctx{ToolRes: []ToolResult{{Name:"weather_query", Plugin:"weather", Result: dirty}}}
var wg sync.WaitGroup
wg.Add(2)
go func(){ defer wg.Done(); kernelExternal(sc, sanitizerStage) }()
go func(){ defer wg.Done(); kernelExternal(sc, weatherStage) }()
wg.Wait()
if s, ok := sc.ToolRes[0].Result.(string); ok && strings.Contains(s, "\x1b[31m") {
polluted++
}
}
fmt.Printf("现网场景复刻:模型调用 weather_querysanitizer+weather 并发跑 AfterToolcall\n")
fmt.Printf(" %d 轮中 %d 轮清洗结果被覆盖 (%.1f%%)\n", rounds, polluted, float64(polluted)/rounds*100)
if polluted > 0 {
fmt.Printf("\n ⚠️ 确认weather 回传的未清洗快照覆盖了 sanitizer 的清洗结果\n")
fmt.Printf(" → 脏数据ANSI 转义)进入 LLM 上下文\n")
}
}

View File

@ -0,0 +1,62 @@
//go:build ignore
package main
/*
#cgo LDFLAGS: -ldl
#include <dlfcn.h>
#include <stdlib.h>
typedef void (*fn)(void);
static void call(void* f){ ((fn)f)(); }
*/
import "C"
import (
"fmt"
"os"
"os/exec"
"runtime"
"time"
"unsafe"
)
func threads() int { e,_ := os.ReadDir("/proc/self/task"); return len(e) }
func main() {
fmt.Println("=== A. cgo 模型:插件死循环,超时后能回收吗? ===")
p := C.CString("./hang.so"); h := C.dlopen(p, C.RTLD_NOW); C.free(unsafe.Pointer(p))
n := C.CString("hang_forever"); f := C.dlsym(h, n); C.free(unsafe.Pointer(n))
base := threads()
fmt.Printf(" 基线: goroutines=%d threads=%d\n", runtime.NumGoroutine(), base)
for i := 1; i <= 3; i++ {
done := make(chan string, 1)
go func() { C.call(f); done <- "ok" }() // 模拟 executeToolCallInner
select {
case <-done:
case <-time.After(600 * time.Millisecond): // 缩短的"60s 超时"
}
time.Sleep(200 * time.Millisecond)
fmt.Printf(" 第 %d 次超时后: goroutines=%d threads=%d (+%d)\n",
i, runtime.NumGoroutine(), threads(), threads()-base)
}
fmt.Println(" ❌ 每次超时永久泄漏 1 goroutine + 1 OS 线程cgo 调用不可中断)")
fmt.Println("\n=== B. 子进程模型:同样死循环,可强杀 ===")
base2 := threads()
for i := 1; i <= 3; i++ {
cmd := exec.Command("sleep", "3600")
cmd.Start()
done := make(chan error, 1)
go func() { done <- cmd.Wait() }()
select {
case <-done:
case <-time.After(300 * time.Millisecond):
cmd.Process.Kill() // ← 可强制终止
<-done
}
fmt.Printf(" 第 %d 次超时+Kill 后: goroutines=%d threads=%d (+%d)\n",
i, runtime.NumGoroutine(), threads(), threads()-base2)
}
fmt.Println(" ✅ 零泄漏进程被杀OS 回收全部资源")
}

View File

@ -0,0 +1,43 @@
//go:build ignore
package main
/*
#cgo LDFLAGS: -ldl
#include <dlfcn.h>
#include <stdlib.h>
typedef void (*fn)(void);
static void call(void* f){ ((fn)f)(); }
*/
import "C"
import (
"fmt"
"os"
"runtime"
"time"
"unsafe"
)
func threads() int { e,_ := os.ReadDir("/proc/self/task"); return len(e) }
func main() {
p := C.CString("./hang.so"); h := C.dlopen(p, C.RTLD_NOW); C.free(unsafe.Pointer(p))
n := C.CString("hang_forever"); f := C.dlsym(h, n); C.free(unsafe.Pointer(n))
base := threads()
fmt.Printf("基线 threads=%d goroutines=%d\n\n", base, runtime.NumGoroutine())
for i := 1; i <= 20; i++ {
done := make(chan string, 1)
go func() { C.call(f); done <- "ok" }()
select {
case <-done:
case <-time.After(120 * time.Millisecond):
}
if i%5 == 0 {
fmt.Printf(" %2d 次卡死调用后: goroutines=%2d threads=%2d (+%d)\n",
i, runtime.NumGoroutine(), threads(), threads()-base)
}
}
fmt.Printf("\n结论: 20 次超时 → 泄漏 %d goroutine, %d OS 线程\n",
runtime.NumGoroutine()-1, threads()-base)
fmt.Println("每个卡在 cgo 里的 goroutine 独占一个 MOS 线程),无法被抢占或回收")
}

View File

@ -0,0 +1,2 @@
#include <unistd.h>
void hang_forever(void) { while(1) sleep(1); }

View File

@ -0,0 +1,111 @@
# 插件架构评估实验
[`../../架构迁移评估.md`](../../架构迁移评估.md) 中所有数字的来源。
**18 项实验,一键复跑**,用于复核结论或在改动后验证回归。
```bash
./run.sh # 跑全部(约 3-5 分钟)
./run.sh 12 13 # 只跑指定实验
./run.sh 1 1c # dlclose/NODELETE 组
```
依赖:`go >= 1.21``gcc`、Linux用到 `eventfd`/`memfd_create`/`dlopen`)。
脚本在 `mktemp -d` 里构建,**不污染主仓 `go.mod`**;实验源码均带 `//go:build ignore`
拉取 `golang.org/x/sys` 需要网络(实验 1/2/4/8/10。本机走 clash
```bash
export HTTPS_PROXY=http://127.0.0.1:7890 HTTP_PROXY=http://127.0.0.1:7890
```
## 目录
| 目录 | 主题 | 对应章节 |
|---|---|---|
| `01-dlclose-nodelete/` | `dlclose``DF_1_NODELETE` 是 no-op | 1.1 / 1.2 |
| `02-feasibility/` | 新架构可行性 11 项 | 第七章 |
| `03-lost-update/` | 副本模型的 lost update | 8.4 / 8.6 |
| `04-cgo-uninterruptible/` | cgo 调用不可中断 | 9.3 |
## 实验清单与最近一次实测结果
复跑于 2026-08-31go1.25.12 linux/amd64192.168.2.6012 核)。
### 01 组dlclose / NODELETE
| # | 实验 | 结论 |
|---|---|---|
| 1a | Go 宿主经纯 C shim 加载/卸载第三层 `.so` | 纯 C 目标可卸载Go c-shared 目标仍不可 |
| 1b | `/proc/self/maps` 段数验证 | 纯 C: 5→**0**真卸载Go c-shared: 5→**5** |
| 1c | 版本化路径 dlopen | handle 不同,`ver=v2` 生效(方案可行但泄漏,已否决) |
**关键**`DF_1_NODELETE` 属于**被卸载对象自身**的 ELF 属性,
与谁调用 `dlopen` 无关——套任何层数的 C 中间件都绕不过去。
### 02 组:新架构可行性
| # | 实验 | 最近结果 |
|---|---|---|
| 1 | eventfd 是否走 Go netpoller | 200 goroutine 阻塞 → 线程 **+0~1** ✅ |
| 2 | 跨进程 eventfd + 偏移解引用 | 父子 mmap 基址不同偏移仍正确post **10.9 µs** |
| 3 | 锁仲裁 RPC 往返成本 | **19.4 µs/次**20000 次) |
| 4 | post-and-forget vs 同步 Publish | 5.07s → 2.29ms**2218x** |
| 5 | 17 子进程常驻开销 | **29.1MB RSS / 12.9MB PSS**84 线程 |
| 6 | 子进程崩溃隔离 | 退出码 **2**EOF **2.5ms** 感知,宿主存活 |
| 7 | 子进程热重载 | 同路径替换二进制 → v1→v2 立即生效 |
| 8 | **跨进程并发改写 StageContext** | 5 进程 × 300 轮,**零丢失零撕裂** |
| 9 | 持锁进程崩溃自愈 | 无死锁,**无需 robust mutex** |
| 10 | 二进制零拷贝 | 100KB/1MB/5MB → **14-22x**,体积 100% |
| 11 | 工具调用 RPC 延迟 | p50 **19.6 µs**,占 LLM 往返 0.00065% |
### 03 组:副本模型缺陷
| # | 实验 | 最近结果 |
|---|---|---|
| 12 | 副本模型 lost update 率 | 内置 **0%** vs 外部 **35.8~36.8%** |
| 13 | 现网 sanitizer+weather 冲突 | **1.6~4.3%** 清洗结果被覆盖 |
**实验 12 的对照设计是重点**:两组用**完全相同的并发扇出**
`stages.go:124``go func` + `wg.Wait()`),唯一差异是
「共享同一 `*StageContext`」vs「快照-副本-写回」。
内置组 0% 证明**并发扇出这个原始设计是正确的**
副本组 36% 证明**跨 C ABI 边界后锁语义失效**才是缺陷所在。
不要据此得出"应该取消并发"的结论。
⚠️ **13 的比率随机器负载波动**(观测区间 1.6%~4.3%)——它取决于两个插件
handler 的实际执行耗时比。文档正文引用 1.6% 是首次测量值,
**应理解为「量级在百分之几」而非精确常数**
### 04 组cgo 不可中断
| # | 实验 | 最近结果 |
|---|---|---|
| 14a | cgo 死循环 vs 子进程 Kill | cgo 泄漏;子进程 **零泄漏** |
| 14b | 泄漏增长曲线20 次) | 泄漏 **20 goroutine / 18 OS 线程**,线性 |
## 复跑时的注意事项
**结果会有波动,以下属正常**
- 实验 12/13 的丢失率随调度波动12 稳定在 35~37%13 在 1.6~4.3%
- 实验 1 的线程增长为 0 或 1取决于 netpoller 线程是否已存在)
- 实验 10 的加速比 14~22x受 CPU 缓存状态影响)
- 实验 5 的 PSS 受同机其他 Go 进程影响(共享页计算)
**结果不应变的**(若变了说明环境或结论有问题):
- 实验 1b 中纯 C `.so` 的段数必须归 **0**Go c-shared 必须**不归零**
- 实验 8 的「总字符数 == 最终长度」必须成立(零丢失)
- 实验 9 必须无死锁
- 实验 12 的内置模型必须 **0%**
- 实验 14b 的泄漏必须**线性增长**
## 已知限制
- 实验 8 的 arena 未实现压实64KB 用尽即停止写入(写入次数 < 5×300 属预期
见评估文档 3.3
- 实验 12/13 **链路复刻**而非直接调用生产代码
证明的是副本模型这一机制存在缺陷不能替代对 `sanitizer`/`weather`
的真实行为回归测试
- 实验 5 的插件是最小 stdio loop2.68MB真实插件 qq 7.5MB开销更高
- Windows 环境9.2 Windows DLL 缺陷**未经实测**仅代码阅读

View File

@ -0,0 +1,127 @@
#!/usr/bin/env bash
# 插件架构评估实验 —— 一键复跑
# 用法: ./run.sh [实验编号...] 例: ./run.sh 12 13 留空跑全部
# 依赖: go >= 1.21, gcc, Linux (eventfd/memfd/dlopen)
set -uo pipefail
cd "$(dirname "$0")"
ROOT=$(pwd)
PASS=0; FAIL=0
need() { command -v "$1" >/dev/null || { echo "缺少依赖: $1"; exit 1; }; }
need go; need gcc
# 统一的临时 module 环境(避免污染主仓 go.mod
WORK=$(mktemp -d); trap 'rm -rf "$WORK"' EXIT
banner() { echo; echo "════════ $* ════════"; }
# x/sys 只有 exp1/2/4/8/10 需要
prep_xsys() {
cat > "$1/go.mod" <<EOF
module exp
go 1.21
require golang.org/x/sys v0.20.0
EOF
(cd "$1" && GOFLAGS=-mod=mod go get golang.org/x/sys@v0.20.0 >/dev/null 2>&1)
}
prep_plain() { printf 'module exp\ngo 1.21\n' > "$1/go.mod"; }
run_go() { # <目录> <说明>
if (cd "$1" && go run . 2>&1); then PASS=$((PASS+1)); else echo " ❌ 失败: $2"; FAIL=$((FAIL+1)); fi
}
SEL="${*:-all}"
sel() { [ "$SEL" = "all" ] && return 0; case " $SEL " in *" $1 "*) return 0;; esac; return 1; }
# ── 01: dlclose / NODELETE ────────────────────────────────
if sel 1; then
banner "实验 1 组: dlclose 对 DF_1_NODELETE 是 no-op"
W=$WORK/e01; mkdir -p $W; cp 01-dlclose-nodelete/*.c $W/
gcc -shared -fPIC -o $W/probe_v1.so $W/probe_v1.c
gcc -shared -fPIC -o $W/probe_v2.so $W/probe_v2.c
gcc -shared -fPIC -o $W/shim.so $W/shim.c
cp $W/probe_v1.so $W/probe.so
for e in exp01a exp01b; do
mkdir -p $W/$e; cp 01-dlclose-nodelete/$e/main.go $W/$e/
sed -i '/^\/\/go:build ignore$/d' $W/$e/main.go; prep_plain $W/$e
(cd $W/$e && go build -o ../$e.bin . 2>&1 | head -3)
done
echo "--- 01a: Go 宿主经 C shim 加载/卸载纯 C so ---"
(cd $W && ./exp01a.bin) && PASS=$((PASS+1)) || FAIL=$((FAIL+1))
echo "--- 01b: /proc/self/maps 段数验证(纯 C 归零Go c-shared 不归零)---"
(cd $W && ./exp01b.bin) && PASS=$((PASS+1)) || FAIL=$((FAIL+1))
fi
# ── 01c: 版本化路径(需要两个真 Go c-shared────────────────
if sel 1c; then
banner "实验 1c: 版本化路径 dlopen 可加载新代码"
W=$WORK/e01c; mkdir -p $W/{v1,v2,host}
for V in v1 v2; do
cat > $W/$V/main.go <<EOF
package main
import "C"
//export lib_version
func lib_version() *C.char { return C.CString("$V-CODE") }
func main() {}
EOF
printf 'module gl%s\ngo 1.21\n' $V > $W/$V/go.mod
(cd $W/$V && go build -buildmode=c-shared -o ../gl$V.so . 2>&1|head -3)
done
cp 01-dlclose-nodelete/exp01c/main.go $W/host/
sed -i '/^\/\/go:build ignore$/d' $W/host/main.go; prep_plain $W/host
(cd $W/host && go build -o ../h.bin .) && (cd $W && ./h.bin) && PASS=$((PASS+1)) || FAIL=$((FAIL+1))
fi
# ── 02: 可行性 1-11 ───────────────────────────────────────
declare -A XSYS=([1]=1 [2]=1 [4]=1 [8]=1 [10]=1)
for n in 1 2 3 4 5 6 7 8 9 10 11; do
sel $n || continue
banner "实验 $n"
W=$WORK/f$n; mkdir -p $W
case $n in
1) cp 02-feasibility/exp1_eventfd.go $W/main.go ;;
2) cp 02-feasibility/exp2_parent.go $W/main.go; cp 02-feasibility/exp2_child.go $W/ ;;
3) cp 02-feasibility/exp3_parent.go $W/main.go; cp 02-feasibility/exp3_child.go $W/ ;;
4) cp 02-feasibility/exp4.go $W/main.go ;;
5) cp 02-feasibility/exp5b.go $W/main.go; cp 02-feasibility/exp5_plugin.go $W/ ;;
6) cp 02-feasibility/exp6.go $W/main.go; cp 02-feasibility/exp6_crash.go $W/ ;;
7) cp 02-feasibility/exp7.go $W/main.go ;;
8) cp 02-feasibility/exp8.go $W/main.go; cp 02-feasibility/exp8_worker.go $W/ ;;
9) cp 02-feasibility/exp9.go $W/main.go; cp 02-feasibility/exp9_worker.go $W/ ;;
10) cp 02-feasibility/exp10.go $W/main.go ;;
11) cp 02-feasibility/exp11.go $W/main.go; cp 02-feasibility/exp11_plug.go $W/ ;;
esac
# 去掉 main.go 的 build ignore它是入口
sed -i '/^\/\/go:build ignore$/d' $W/main.go
if [ "${XSYS[$n]:-}" = "1" ]; then prep_xsys $W; else prep_plain $W; fi
# 需要预编译的辅助二进制
case $n in
5) (cd $W && go build -o plugbin exp5_plugin.go 2>&1|head -3) ;;
6) (cd $W && go build -o crashbin exp6_crash.go 2>&1|head -3) ;;
11) (cd $W && go build -o plug11 exp11_plug.go 2>&1|head -3) ;;
esac
run_go $W "实验 $n"
done
# ── 03: lost update ───────────────────────────────────────
for e in 12 13; do
sel $e || continue
banner "实验 $e: 副本模型 lost update"
W=$WORK/l$e; mkdir -p $W
cp 03-lost-update/exp$e/main.go $W/; sed -i '/^\/\/go:build ignore$/d' $W/main.go
prep_plain $W; run_go $W "实验 $e"
done
# ── 04: cgo 不可中断 ──────────────────────────────────────
for e in 14a 14b; do
sel 14 || sel $e || continue
banner "实验 $e: cgo 调用不可中断"
W=$WORK/c$e; mkdir -p $W
cp 04-cgo-uninterruptible/hang.c $W/
gcc -shared -fPIC -o $W/hang.so $W/hang.c
cp 04-cgo-uninterruptible/exp$e/main.go $W/; sed -i '/^\/\/go:build ignore$/d' $W/main.go
prep_plain $W; run_go $W "实验 $e"
done
banner "汇总: 通过 $PASS, 失败 $FAIL"
[ $FAIL -eq 0 ]