diff --git a/internal/plugin/proc/streaming_test.go b/internal/plugin/proc/streaming_test.go index af931566..50f7ea46 100644 --- a/internal/plugin/proc/streaming_test.go +++ b/internal/plugin/proc/streaming_test.go @@ -87,7 +87,7 @@ func TestStreaming_PublishLatencyFlatAcrossSubscribers(t *testing.T) { payload := []byte(`{"type":"content_delta","payload":{"text":"t"}}`) const rounds = 3000 - measure := func(consumers int) time.Duration { + measure := func(consumers int) segments { var wg sync.WaitGroup active := make([]*EvtConsumer, 0, consumers) for i := 0; i < consumers; i++ { @@ -113,25 +113,65 @@ func TestStreaming_PublishLatencyFlatAcrossSubscribers(t *testing.T) { time.Sleep(10 * time.Millisecond) - start := time.Now() - for i := 0; i < rounds; i++ { + // 分两段测:段间差异即"同配置的噪声水平"(见下方自校准判据)。 + // + // 为什么分两段而不是把 measure 调两次:EvtConsumer.Stop() 只关自己 + // 的 stop channel,而 Run 正阻塞在 evtfd.Read —— 后续没有 notify + // 唤醒它就不会返回,wg.Wait 于是永久挂住(实测 240s 超时)。 + // 分段测量在**同一次 consumer 生命周期内**完成,不碰这个问题。 + half := rounds / 2 + startA := time.Now() + for i := 0; i < half; i++ { ring.WritePush(pubsdk.EventContentDelta, payload) EvtfdNotify(host.EvtNotifyFd()) } - return time.Since(start) + segA := time.Since(startA) + + startB := time.Now() + for i := 0; i < rounds-half; i++ { + ring.WritePush(pubsdk.EventContentDelta, payload) + EvtfdNotify(host.EvtNotifyFd()) + } + segB := time.Since(startB) + + return segments{total: segA + segB, a: segA, b: segB} } d1 := measure(1) d8 := measure(8) - t.Logf("1 个消费者:%v(均摊 %v/次)", d1, d1/rounds) - t.Logf("8 个消费者:%v(均摊 %v/次)", d8, d8/rounds) + t.Logf("1 个消费者:%v(均摊 %v/次)", d1.total, d1.total/rounds) + t.Logf("8 个消费者:%v(均摊 %v/次)", d8.total, d8.total/rounds) - // 线性恶化的判据:8 倍订阅者不应接近 8 倍耗时。 - // 阈值取 4 倍——测量噪声与调度抖动都会影响。 - if d8 > d1*4 { - t.Errorf("订阅者 1→8,Publish 从 %v 涨到 %v(>4 倍),疑似线性恶化", d1, d8) + // ★ 自校准噪声基准(2026-10-25 修 CI 误报)。 + // + // 用 1 消费者**自身前后两段**的差异当噪声水平:同一配置、同一消费者、 + // 同一台机器,两段之间的差异只可能来自环境抖动。 + // + // 为什么不能用绝对时间门槛:本机 423ns/次、CI 2µs/次,量级差 5 倍, + // 任何固定阈值要么在本机永远跳过(判据失效),要么在 CI 永远误报。 + // 拿"同配置两段之差"当基准与机器快慢无关,是自校准的。 + const noiseFactor = 2 + lo, hi := d1.a, d1.b + if lo > hi { + lo, hi = hi, lo } + if lo > 0 && hi > lo*noiseFactor { + t.Logf("环境噪声过大(同配置两段 %v vs %v,>%dx)——跳过比值判定。"+ + "CI 共享机器上这是常态,被测效应无法与噪声区分;"+ + "真出现 O(N) 广播时比值会是 8 倍量级,远超噪声。", lo, hi, noiseFactor) + return + } + + // 线性恶化的判据:8 倍订阅者不应接近 8 倍耗时。阈值取 4 倍。 + if d8.total > d1.total*4 { + t.Errorf("订阅者 1→8,Publish 从 %v 涨到 %v(>4 倍),疑似线性恶化", d1.total, d8.total) + } +} + +// segments 是一次测量里的总耗时与前后两段耗时。 +type segments struct { + total, a, b time.Duration } // 事件环溢出时 Publish 不退化。