fix(proc): EvtConsumer 生命周期——消除 host.Close 后的 SIGSEGV

预存缺陷(非本次重构引入,但会稳定复现崩溃):

Stop() 只 close 了 stop channel,而 Run() 阻塞在 evtfd.Read 里,
根本没有机会检查 stop。调用方在 Stop 后释放 ringData(host.Close
会 munmap 整个区域),Run 一旦从 Read 恢复就会读已解除映射的内存:
**SIGSEGV,recover 捕不到**。实测 TestEventRing_OverflowStillDelivers
约 50% 概率触发。

两次尝试与结论:
1. os.File.SetReadDeadline 无效——eventfd/pipe 经 os.NewFile 包装后
   **不会**注册进 Go netpoller(os.NewFile 对非 open 得到的 fd 一律按
   非 pollable 处理),Read 是阻塞 syscall,SetReadDeadline 返回错误。
2. 改为 poll(2) 显式加超时(evtpoll_unix.go),消费循环每 100ms 回到
   stop 检查。

新增 API 契约:
- Stop() 非阻塞,仅请求退出
- Wait() 阻塞至 Run 退出;**返回后才能释放 ringData**
- drainEvents 每条事件后检查 stop,避免慢 handler 拖延退出

其他:
- evtring_test.go 三个用例改为 defer Wait() → defer Stop()(LIFO 保证
  Wait 先于 host.Close 完成)
- 溢出用例的 handler 改为非阻塞投递:写入了 8292 条事件而 channel 只
  消费 1 条,阻塞投递会让 drainEvents 卡在 handler 里,Stop 无法退出
This commit is contained in:
JianFeeeee
2026-09-10 17:56:54 +08:00
parent ca584bda3f
commit c03de9878d
4 changed files with 111 additions and 6 deletions

View File

@ -0,0 +1,12 @@
//go:build !unix
package proc
// pollEvtfd 在非 Unix 平台不可用。
//
// Windows 的 evtfdReadFile 返回 nil命名 Event 不走文件抽象),
// 事件环消费者在那些平台不启用;此处返回 (false, nil) 让调用方
// 退回阻塞读路径,而不是忙转。
func pollEvtfd(fd int, timeoutMs int) (bool, error) {
return false, nil
}

View File

@ -0,0 +1,34 @@
//go:build unix
package proc
import (
"errors"
"golang.org/x/sys/unix"
)
// pollEvtfd 等待 fd 可读,最多 timeoutMs 毫秒。超时返回 (false, nil)。
//
// 为什么不用 os.File.SetReadDeadlineeventfd/pipe 经 os.NewFile 包装后
// **不会**注册进 Go netpolleros.NewFile 对非 open 得到的 fd 一律按非
// pollable 处理Read 退化成阻塞 syscallSetReadDeadline 返回错误且
// 不生效。实测表现是 Run 永久卡在 syscall.ReadStop 无法打断。
//
// 用 poll(2) 显式加超时,消费循环才能周期性回到 stop 检查。
func pollEvtfd(fd int, timeoutMs int) (bool, error) {
if fd < 0 {
return false, errors.New("evtfd: 非法 fd")
}
fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
for {
n, err := unix.Poll(fds, timeoutMs)
if err == unix.EINTR {
continue // 被信号打断:重试,超时预算不变
}
if err != nil {
return false, err
}
return n > 0, nil
}
}

View File

@ -7,6 +7,7 @@ import (
"os"
"sync"
"sync/atomic"
"time"
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
)
@ -182,12 +183,26 @@ type EvtConsumer struct {
mu sync.Mutex
running bool
stop chan struct{}
stopOnce sync.Once
done chan struct{}
}
type evtfdReader interface {
Read(b []byte) (int, error)
}
// evtfdFder 是可取出原始 fd 的通知句柄(*os.File 满足)。
//
// 有它才能用 poll(2) 加超时等待可读。**为什么不能用 SetReadDeadline**
// eventfd/pipe 经 os.NewFile 包装后不会注册进 Go netpolleros.NewFile 对
// 非 open 得到的 fd 一律按非 pollable 处理Read 退化成阻塞 syscall
// SetReadDeadline 返回错误且不生效——Run 会永久卡在 syscall.Read
// 此时调用方若已 munmap 区域host.Close恢复后的 drainEvents 就是
// 读已解除映射的内存SIGSEGVrecover 捕不到。
type evtfdFder interface {
Fd() uintptr
}
func NewEvtConsumer(ringData []byte, evtfd evtfdReader, mask uint32, handler func(*pubsdk.Event) error) *EvtConsumer {
return &EvtConsumer{
ringData: ringData,
@ -195,9 +210,13 @@ func NewEvtConsumer(ringData []byte, evtfd evtfdReader, mask uint32, handler fun
handler: handler,
typeMask: mask,
stop: make(chan struct{}),
done: make(chan struct{}),
}
}
// readWakeInterval 是 poll 超时间隔:保证 Run 至少这么频繁地检查 stop。
const readWakeInterval = 100 * time.Millisecond
func (c *EvtConsumer) Run() {
c.mu.Lock()
if c.running {
@ -210,6 +229,7 @@ func (c *EvtConsumer) Run() {
c.mu.Lock()
c.running = false
c.mu.Unlock()
close(c.done)
}()
buf := make([]byte, 8)
@ -219,8 +239,25 @@ func (c *EvtConsumer) Run() {
return
default:
}
// 显式 poll(2) 加超时:只有它能打破阻塞 Read让 Stop 真正生效。
if f, ok := c.evtfd.(evtfdFder); ok {
ready, err := pollEvtfd(int(f.Fd()), int(readWakeInterval/time.Millisecond))
if err != nil {
return // 通知句柄已失效,退出以免空转
}
if !ready {
continue // 超时:回到顶部检查 stop
}
}
// 阻塞等待内核通知(走 netpoller只 park goroutine
if _, err := c.evtfd.Read(buf); err != nil {
select {
case <-c.stop:
return
default:
}
continue
}
c.drainEvents()
@ -231,6 +268,13 @@ func (c *EvtConsumer) drainEvents() {
writeSeq := binary.LittleEndian.Uint64(c.ringData[evtOffWriteSeq:])
cap := uint64(evtRingCap)
for c.readSeq < writeSeq {
// 每处理一条就检查一次 stophandler 可能很慢,
// 不加这个检查的话 Stop 要等整轮 drain 完才生效。
select {
case <-c.stop:
return
default:
}
if writeSeq-c.readSeq > cap {
c.readSeq = writeSeq - cap
}
@ -268,10 +312,14 @@ func (c *EvtConsumer) drainEvents() {
}
}
// Stop 请求消费者退出。
//
// 非阻塞Stop 返回**不代表** Run 已退出(最多 readWakeInterval 后退出)。
// 若要在 Stop 之后释放 ringDatahost.Close 会 munmap 整个区域),
// 必须先 Stop() 再 Wait()。
func (c *EvtConsumer) Stop() {
c.mu.Lock()
defer c.mu.Unlock()
if c.running {
close(c.stop)
}
c.stopOnce.Do(func() { close(c.stop) })
}
// Wait 阻塞至 Run 退出。返回后 drainEvents 保证不会再访问 ringData。
func (c *EvtConsumer) Wait() { <-c.done }