fix(memory): 面板调优引入的两处运行期写入面收口（-race 修前红 → 修后绿）
一、审查档（preinsert.go / preinsert_gate.go）：面板调优会在**运行期**改 cfg 的 两个字段（DisableIngestGate / TempTTL），而读这两个字段的两处不持锁 （gateEnabled 在同步入库闸上、deferDeadline 在第五态 defer 上）。 - 读侧改走新的 tuningSnapshot()（r.mu 由 Mutex 换成 RWMutex，读锁下取整份快照）； - **只**收这两个字段的读点：其余读点读的是构造期冻结的字段，不与写方交手， 故刻意不给入库热路径加无谓的锁；注释写明「将来面板再接管别的字段，那些读点 必须与它一起收」； - 锁序：写方持锁期间只做字段赋值、不取任何别的锁（normalize/回落都在取锁之前 跑完）⇒ 不引入新锁序。 二、整理档（tidy_service_tuning.go）：修掉我自己的那条真竞争 —— 原顺序是 「Stop → Start → 写 cfg」，而新起的循环**立刻**读 s.cfg 重算下一拍，与写入并排 （那一拍还可能按旧配置算出来）。改成「Stop → 写 → Start」。 顺带去掉 Stop 的超时分支：超时意味着 wg 的 Wait 仍在飞，紧接着 Start 的 Add 会 踩上 M3 在 tidy_service.go 明文写下的 WaitGroup 复用纪律（-race 下 panic）；`Stop` 期间也不持 runMu，避免与「心跳正要开跑一轮」互相等死。 证据（全部逐条跑过再还原，还原用 diff 逐字校验为 0 差异）： - 审查档：撤回读侧修复 ⇒ 2 处 DATA RACE（写 preinsert_tuning.go 的两个字段 × 读 gateEnabled / deferDeadline）；恢复 ⇒ 绿。 - 整理档：把顺序改回「先起后写」⇒ DATA RACE（写 applyTidyConfig × 读 nextDelay）； 恢复 ⇒ 绿。 - 新增两条只在 -race 下有意义的回归哨兵（tidy_service_tuning_test.go）：审查档真 并发读写；整理档用**真心跳**当读者（不伪造生产不存在的调用方，那个入口在包里 只有心跳 goroutine 会调）。 热路径成本：新增的同步是「一次无竞争 RLock + 整份小结构体快照」，隔离实测约 4ns/次（5e6 次共 22.9ms）；入库闸那条测量用例的批均值在 2.62ms → 2.71~2.81ms 之间波动，落在那条用例自身的运行间噪声里（同一份代码连跑两次，纯写入基线就从 39µs 飘到 60µs）。
This commit is contained in:
AskaEth committed
1 parent
e9e7b7dcb2
commit
a87fbeb494
5 files changed
+193
-70
No files matched your search
@@ -2,10 +2,8 @@ package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
|
||||
)
|
||||
@@ -40,48 +38,59 @@ import (
|
||||
// 调度判定就用它），并把失败如实回给调用方 —— 绝不假装热生效，也绝不让保存白写。
|
||||
//
|
||||
// 并发口径（为什么这样写是安全的）：`s.runMu` 串行化**所有**轮次（心跳轮 `runRound`
|
||||
// 与手动轮 / 试算 `DryRun` 都持它），而两个字段的读者只有这些轮次；因此「持 runMu →
|
||||
// Stop 心跳 → 写 cfg → Start」这一串之后，没有任何 goroutine 会在写入的同时读这两个
|
||||
// 字段。心跳 goroutine 的调度读（`nextDelay` / `dueNow` 读 `s.cfg`）在 Stop 返回后
|
||||
// 已经全部结束，因此也不与写入交手。
|
||||
|
||||
// memoryTidyRescheduleTimeout 是重排心跳时「等正在跑的那一轮跑完」的上限。
|
||||
// 与手动轮 / 试算 `DryRun` 都持它），而它排除的是**轮次**；调度读（`nextDelay` /
|
||||
// `dueNow` 读 `s.cfg`）不受 runMu 保护，因此写 `s.cfg` 必须发生在「没有任何 goroutine
|
||||
// 读它」的窗口里 ⇒ 顺序铁律是 **先停、再写、后起**：
|
||||
//
|
||||
// 它存在的理由：`Stop` 要等 wg 排空（可能正有一轮在扫描几百条记忆）。给上限是为了
|
||||
// 不让一次面板保存把 HTTP 请求挂到天荒地老；超时**如实报错**（值照旧落到运行期），
|
||||
// 绝不静默假装重排成功。
|
||||
const memoryTidyRescheduleTimeout = 5 * time.Second
|
||||
|
||||
// ErrTidyRescheduleTimeout 在重排心跳时没能等正在跑的那一轮排空时返回（值已生效于
|
||||
// 下一次调度判定，只是当前那次等待没有被作废）。
|
||||
var ErrTidyRescheduleTimeout = errors.New("memory/tidy: heartbeat reschedule timed out")
|
||||
// Stop（等到心跳与清扫两只后台件都退出）→ 写 cfg → Start（新循环读到的是新配置）
|
||||
//
|
||||
// 三条都是实测踩出来的，不是推演：
|
||||
// - **不能「先拿 runMu 再 Stop」**：心跳若正卡在 runMu 上准备开跑一轮，Stop 等的就是
|
||||
// 一个被我挡在锁门口的 goroutine ⇒ 活锁到超时；
|
||||
// - **不能「Stop 完先 Start 再写」**：新循环立刻就会读 `s.cfg` 重算下一拍，与写入并排
|
||||
// ⇒ `go test -race` 直接报竞争，而且那一拍还可能按**旧**配置算出来；
|
||||
// - **Stop 不设超时**：超时分支里 Wait 仍在飞，紧接着 `Start` 的 `wg.Add` 会踩上
|
||||
// `tidy_service.go` 里 M3 明文写下的 WaitGroup 复用纪律（交叉调用 Start/Stop
|
||||
// 在 `-race` 下 `panic: sync: WaitGroup is reused before previous Wait has returned`）。
|
||||
// 等它排空的最坏代价是「等正在跑的那一轮跑完」，而这条路径是人点一次保存触发的，
|
||||
// 不持 runMu ⇒ 不会与「心跳要开跑一轮」互相锁死。
|
||||
|
||||
// ApplyPanelTuning 把面板值当场叠到运行期（见文件头三条口径）。
|
||||
//
|
||||
// 返回 nil 表示**值已生效、心跳已按新周期重排**；返回错误时**值同样已生效**，只是
|
||||
// 心跳这次没排成（错误文案里说清是哪一种），调用方（网关）据此如实回 `hot:false`。
|
||||
// 返回 nil 表示**值已生效**（心跳在跑时也已被重排，下一拍按新周期算）；返回错误时
|
||||
// 值**没有**落到运行期（落盘那一步早已完成，调用方据此如实回 `hot:false` + 原因）。
|
||||
//
|
||||
// 顺序与三条由来见文件头：**先停 → 再写 → 后起**。
|
||||
func (s *TidyService) ApplyPanelTuning(base TidyConfig, panel runtimeconfig.MemoryTidyTuning) error {
|
||||
next := tidyPanelTuning(base, panel, s.log)
|
||||
|
||||
// 只把 runMu 拿住 —— 它同时排除「正在跑的一轮」与「刚要开跑的一轮」。注意
|
||||
// 顺序：**先拿 runMu 再 Stop**。反过来的话，心跳若正卡在 runMu 上等开跑，Stop
|
||||
// 会等一个永远等不到退出的 goroutine（它被我抱着锁挡在门外）。
|
||||
s.runMu.Lock()
|
||||
defer s.runMu.Unlock()
|
||||
|
||||
s.mu.Lock()
|
||||
running := s.running
|
||||
unchanged := s.cfg == next
|
||||
s.mu.Unlock()
|
||||
|
||||
if unchanged {
|
||||
// 面板这次只改了别的键（如 ingestGate）：处理配置根本没变 ⇒ 不打扰心跳。
|
||||
s.applyTidyConfigLocked(next)
|
||||
// 值与运行期**完全一致**（面板这次只改了别的键）⇒ 一个字段都不用写。
|
||||
// 刻意**不**走下面那条写路径：写一个「相同的值」也是写，照样会与调度读
|
||||
// 构成竞争，而它一点收益都没有。
|
||||
return nil
|
||||
}
|
||||
|
||||
reschedErr := s.rescheduleTidyHeartbeat()
|
||||
s.applyTidyConfigLocked(next)
|
||||
return reschedErr
|
||||
// 心跳在跑 ⇒ 先让两只后台件（心跳 + 到期清扫）停稳：`s.cfg` 的调度读不持锁，
|
||||
// 写入必须落在一个「没有读者」的窗口里。**Stop 期间不持 runMu**（否则心跳若正
|
||||
// 卡在 runMu 上等开跑，会互相等死）。
|
||||
if running {
|
||||
if err := s.Stop(context.Background()); err != nil {
|
||||
return fmt.Errorf("memory/tidy: stop heartbeat before tuning: %w", err)
|
||||
}
|
||||
}
|
||||
s.applyTidyConfig(next)
|
||||
if running {
|
||||
// 重新起后台件：`Stop` 已经把 running 置为 false，不重启就等于让定时整理
|
||||
// **与到期清扫一起**悄悄停摆。父 ctx 用 Background（本对象没有保存父 ctx 的
|
||||
// 字段，那在 tidy_service.go 里），关停链本来就由组合根显式 `Stop` 收尾。
|
||||
s.Start(context.Background())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// tidyPanelTuning 是「面板 > `.env` > 代码默认」的**重算点**（口径 1）。
|
||||
@@ -101,43 +110,19 @@ func tidyPanelTuning(base TidyConfig, panel runtimeconfig.MemoryTidyTuning, log
|
||||
return normalizeTidyConfig(cfg)
|
||||
}
|
||||
|
||||
// applyTidyConfigLocked 把新配置落到**两个**持有者上（调用方必须已持 runMu，并已让
|
||||
// 心跳停下）：
|
||||
// - `s.cfg` 是调度判据（`intervalAt` / `inQuietHours`）的来源；
|
||||
// applyTidyConfig 把新配置落到**两个**持有者上：
|
||||
// - `s.cfg` 是调度判据（`intervalAt` / `inQuietHours`）的来源（`s.mu` 下写）；
|
||||
// - `s.tidier.cfg` 是每一轮实际用的阈值/上限/自动档开关的来源。
|
||||
//
|
||||
// 两处都要写：只写前者 ⇒ 轮次仍用旧阈值；只写后者 ⇒ 调度仍按旧周期。
|
||||
func (s *TidyService) applyTidyConfigLocked(cfg TidyConfig) {
|
||||
//
|
||||
// 它自己拿 `runMu`（排除并发的一轮：心跳轮与手动轮 / 试算都持它），因为调用方
|
||||
// （ApplyPanelTuning）在 `Stop` 期间**不能**持它。前提仍是「心跳已停」：调用方负责。
|
||||
func (s *TidyService) applyTidyConfig(cfg TidyConfig) {
|
||||
s.runMu.Lock()
|
||||
defer s.runMu.Unlock()
|
||||
s.mu.Lock()
|
||||
s.cfg = cfg
|
||||
s.mu.Unlock()
|
||||
s.tidier.cfg = cfg
|
||||
}
|
||||
|
||||
// rescheduleTidyHeartbeat 把当前那次等待作废，并按**新配置**重新排一次心跳。
|
||||
//
|
||||
// 未在跑（未 Start / 已 Stop）时是空操作。排空超时时**仍把心跳挂回去**：`Stop` 已经
|
||||
// 把 running 置为 false，不重启就等于让定时整理**悄悄停摆**（本项目明令禁止的静默
|
||||
// 失效形态）；旧的那只 goroutine 会在当前这一轮跑完后自行退出。
|
||||
//
|
||||
// 重排用的父 context 是 Background：本对象没有保存父 ctx 的字段（那在
|
||||
// tidy_service.go 里，不在本切片 scope 内），而关闭路径本来就由组合根显式 `Stop`
|
||||
// 收尾（见 wire_services.go），因此不依赖父 ctx 取消。
|
||||
func (s *TidyService) rescheduleTidyHeartbeat() error {
|
||||
s.mu.Lock()
|
||||
running := s.running
|
||||
s.mu.Unlock()
|
||||
if !running {
|
||||
return nil
|
||||
}
|
||||
stopCtx, cancel := context.WithTimeout(context.Background(), memoryTidyRescheduleTimeout)
|
||||
defer cancel()
|
||||
if err := s.Stop(stopCtx); err != nil {
|
||||
s.log.Warn("memory tidy tuning applied, but heartbeat reschedule timed out",
|
||||
"timeout", memoryTidyRescheduleTimeout.String(), "error", err)
|
||||
s.Start(context.Background())
|
||||
return fmt.Errorf("%w: %v", ErrTidyRescheduleTimeout, err)
|
||||
}
|
||||
s.Start(context.Background())
|
||||
return nil
|
||||
}
|
||||
Reference in new issue
Block a user