feat(shm): 统一共享内存区域(§13.1) — 单 memfd 容纳 SuperBlock+StageContext+EvtRing

- unified.go: SuperBlock 布局 + SharedRef 类型
- NewHost(): 两段 memfd 合并为单一 memfd,fd 3 = 统一区域,fd 4 = eventfd
- 子进程侧: testdata 插件从 SuperBlock 解析 ctxOff/evtOff
- streaming 测试: EvtConsumer 协程在 host.Close 前 Stop+Wait 防 SIGSEGV
This commit is contained in:
JianFeeeee
2026-09-10 11:42:22 +08:00
parent 7a328679da
commit ad016e4410
7 changed files with 291 additions and 60 deletions

View File

@ -26,18 +26,26 @@ func TestStreaming_SlowConsumerDoesNotBlockPublish(t *testing.T) {
if err != nil {
t.Fatalf("NewHost: %v", err)
}
defer host.Close()
ring := host.EvtRing()
var consumed atomic.Int64
consumer := NewEvtConsumer(host.EvtData(), host.EvtfdReadFile(), 0,
func(evt *pubsdk.Event) error {
time.Sleep(20 * time.Microsecond) // 刻意的慢订阅者
time.Sleep(20 * time.Microsecond)
consumed.Add(1)
return nil
})
go consumer.Run()
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
consumer.Run()
}()
// defer LIFO:Stop → 关 evtfd 打断 Read → 等协程退出 → Close host
defer host.Close()
defer wg.Wait()
defer host.EvtfdReadFile().Close()
defer consumer.Stop()
const tokens = 5000
@ -71,7 +79,9 @@ func TestStreaming_PublishLatencyFlatAcrossSubscribers(t *testing.T) {
if err != nil {
t.Fatalf("NewHost: %v", err)
}
// defer LIFO:先停消费者再 Close host
defer host.Close()
defer host.EvtfdReadFile().Close()
ring := host.EvtRing()
payload := []byte(`{"type":"content_delta","payload":{"text":"t"}}`)
@ -93,13 +103,14 @@ func TestStreaming_PublishLatencyFlatAcrossSubscribers(t *testing.T) {
cc.Run()
}(c)
}
// defer LIFO:Stop → 等协程退出
defer wg.Wait()
defer func() {
for _, c := range active {
c.Stop()
}
}()
// 让消费者先就位
time.Sleep(10 * time.Millisecond)
start := time.Now()