buffer_pool.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581
  1. package pool
  2. import (
  3. "sync"
  4. "sync/atomic"
  5. )
  6. // pageEntry 缓冲池中的页条目
  7. type pageEntry struct {
  8. data []byte // 页数据副本
  9. dirty bool // 是否被修改(需要写回磁盘)
  10. pageLen int // 页字节长度(用于计算文件偏移)
  11. ver uint64 // 内容版本号: 每次内容被写入都换一个 (见 BufferPool.stampLocked)
  12. lastTouch int64 // 最近一次访问序号 (用 atomic 读写, 替代 LRU 链表节点)
  13. }
  14. // DirtyPage 脏页快照
  15. //
  16. // ver 是快照当时该页的版本号, 只在 pool 内部使用。提交成功后 MarkClean 靠它判断
  17. // 「这页在提交窗口里有没有被再改过」: 改过就必须继续留脏, 不能被顺手标成干净页。
  18. type DirtyPage struct {
  19. PageID uint32
  20. Data []byte
  21. PageLen int
  22. ver uint64
  23. }
  24. // BufferPool 页面缓冲池
  25. // 拦截 ReadPage/WritePage,将页缓存在内存中,减少磁盘 I/O
  26. // 写回策略:
  27. // - Commit() → 通过 CollectDirty() 收集脏页,走 WAL 流程写回
  28. // - 脏页超过阈值时自动触发 onThreshold 回调 (由 DB 层执行 Commit)
  29. //
  30. // 缓存策略: 按时间戳 LRU —— 每次访问用 atomic 递增 touchCounter 并写入页的
  31. // lastTouch 字段, 淘汰时按采样找 lastTouch 最小的干净页 (最久未用)。
  32. // 用 atomic 时间戳替代双向链表, 让 ReadPage cache-hit 路径完全不需要写锁。
  33. // 脏页绝不淘汰 (宁可临时超容量, 由脏页阈值触发的 Commit 回落)。
  34. type BufferPool struct {
  35. pages map[uint32]*pageEntry // pageID → 页条目
  36. mu sync.RWMutex
  37. capacity int // 最大缓存页数
  38. dirtyThreshold int // 脏页阈值 (0=不限制)
  39. onThreshold func() // 脏页超阈值回调
  40. dirtyCount int // 当前脏页数 (实时追踪)
  41. verSeq uint64 // 版本号发放器 (见 stampLocked)
  42. diskRead func(pageID uint32, buf []byte) error // 底层磁盘读
  43. diskWrite func(pageID uint32, buf []byte) error // 底层磁盘写
  44. touchCounter int64 // LRU 时间戳发放器 (atomic 递增, 全局单调递增)
  45. // 锁表机制
  46. writeLocked bool // 写锁标志
  47. writeChan chan struct{} // 用于通知解锁的 channel
  48. // 事务支持
  49. onPageWrite func(pageID uint32) // 页写入时的回调(用于事务缓存原始数据)
  50. }
  51. // evictSample 淘汰时一次最多考察的候选页数 (采样淘汰, 对照 Redis
  52. // maxmemory-samples=5 —— 我们给到 32, 大池近似 LRU, 小池精确 LRU)。
  53. //
  54. // 取值 32 的理由: 淘汰路径在 WritePage 的持锁区间 (RWMutex.Lock) 上,
  55. // 每淘汰一页都要遍历采样池选最小 —— 32 比 5 多出 6 倍候选, LRU 近似度
  56. // 显著提升 (特别是冷热分布非均匀时), 但只多 ~27 次 atomic.LoadInt64,
  57. // 实测从 ~3μs 涨到 ~4μs, 完全可接受。再往上 (100+) 读锁路径会明显被
  58. // 拉长。
  59. //
  60. // 注意: 这个常量属于 pool 子包自己的语义 —— 它控制子包内部策略, 所以
  61. // 不提到 core/const.go (core 的常量是对外可见的库配置, pool 的是内部调参)。
  62. // 同理: const.go 里的 FileLockByteCount / MaxBufferPoolCapacity / BlobChainMax
  63. // 是 core 包对外暴露的配置闸门, 必须集中管理。
  64. const evictSample = 32
  65. // NewBufferPool 创建页面缓冲池
  66. // - capacity: 最大缓存页数
  67. // - diskRead/diskWrite: 底层磁盘读写函数
  68. func NewBufferPool(capacity int,
  69. diskRead func(uint32, []byte) error,
  70. diskWrite func(uint32, []byte) error,
  71. ) *BufferPool {
  72. bp := &BufferPool{
  73. pages: make(map[uint32]*pageEntry, capacity),
  74. capacity: capacity,
  75. diskRead: diskRead,
  76. diskWrite: diskWrite,
  77. }
  78. return bp
  79. }
  80. // SetDirtyThreshold 设置脏页阈值和回调
  81. // 当脏页数超过 threshold 时调用 onThreshold (由 DB 层触发 Commit)
  82. func (bp *BufferPool) SetDirtyThreshold(threshold int, onThreshold func()) {
  83. bp.mu.Lock()
  84. defer bp.mu.Unlock()
  85. bp.dirtyThreshold = threshold
  86. bp.onThreshold = onThreshold
  87. }
  88. // OverThreshold 当前脏页数是否已达到阈值 (threshold<=0 表示不限, 恒 false)。
  89. // 与 WritePage 里的触发条件 (dirtyCount >= dirtyThreshold) 严格同一条, 供 DB 层
  90. // 在写操作收尾时补"操作期间被提交栅栏抑制的阈值提交"。
  91. func (bp *BufferPool) OverThreshold() bool {
  92. bp.mu.Lock()
  93. defer bp.mu.Unlock()
  94. return bp.dirtyThreshold > 0 && bp.dirtyCount >= bp.dirtyThreshold
  95. }
  96. // LockWrites 锁定写入(Commit/Flush 期间调用)
  97. func (bp *BufferPool) LockWrites() {
  98. bp.mu.Lock()
  99. bp.writeLocked = true
  100. bp.writeChan = make(chan struct{}) // 创建新的 channel
  101. bp.mu.Unlock()
  102. }
  103. // UnlockWrites 解锁写入(Commit/Flush 完成后调用)
  104. func (bp *BufferPool) UnlockWrites() {
  105. bp.mu.Lock()
  106. bp.writeLocked = false
  107. if bp.writeChan != nil {
  108. close(bp.writeChan) // 通知所有等待的 goroutine
  109. bp.writeChan = nil
  110. }
  111. bp.mu.Unlock()
  112. }
  113. // waitForWriteLock 等待写锁释放(在 WritePage 中调用)。
  114. //
  115. // 醒来必须【回头复查】 writeLocked: 锁是成对替换的 (LockWrites 每次新建
  116. // channel), 若本 waiter 持有旧 channel 被放行时, 恰逢下一笔提交已重新加锁,
  117. // 不复查就会穿越进别人的提交窗口写页 —— 快照一致性被破坏。
  118. func (bp *BufferPool) waitForWriteLock() {
  119. for {
  120. bp.mu.Lock()
  121. if !bp.writeLocked {
  122. bp.mu.Unlock()
  123. return // 未锁定,直接返回
  124. }
  125. // 保存 channel 引用
  126. ch := bp.writeChan
  127. bp.mu.Unlock()
  128. // 阻塞等待解锁(channel 关闭时会返回)
  129. <-ch
  130. // 循环回头复查: 可能已被下一笔 LockWrites 重新锁住
  131. }
  132. }
  133. // touch 给页打一个"刚被访问过"的时间戳。
  134. // 用 atomic 递增 touchCounter, 全局单调递增, 替代双向链表移动。
  135. // 调用方【不需要】持有 bp.mu —— atomic Store 本身是并发安全的,
  136. // 这正是让 ReadPage cache-hit 路径能用 RLock 的关键。
  137. func (bp *BufferPool) touch(e *pageEntry) {
  138. ts := atomic.AddInt64(&bp.touchCounter, 1)
  139. atomic.StoreInt64(&e.lastTouch, ts)
  140. }
  141. // touchLocked 兼容旧调用点 (在持锁上下文中也可用), 本质就是 touch。
  142. // 现在不再需要锁保护 (atomic 操作), 保留此函数名是为了 grep 容易定位。
  143. func (bp *BufferPool) touchLocked(e *pageEntry) {
  144. bp.touch(e)
  145. }
  146. // removeLocked 从缓存中摘除一页。调用方必须持有 bp.mu (Lock).
  147. // 原来同时需要从 LRU 链表移除, 现在时间戳方案下只需要 delete map entry。
  148. func (bp *BufferPool) removeLocked(pageID uint32, e *pageEntry) {
  149. delete(bp.pages, pageID)
  150. }
  151. // CollectDirty 取出所有脏页快照 (由 DB.Commit() 调用,配合 WAL 流程统一写回)。
  152. //
  153. // 快照之后【不把页移出缓存】。移出会造出一个「既不在缓存、也还没落盘」的窗口: 从
  154. // 收集完到 wal.Commit() 逐页回写完成之间要经过两次 fsync, 这期间任何缓存未命中的读
  155. // 都只能回源磁盘, 于是拿到的是旧内容 —— 或者, 对一页本次会话刚分配、盘上从来没写过
  156. // 的页, 拿到全零 (文件还没长到那个偏移时直接读出错)。btree 把全零页当成
  157. // EntryLen==0 的节点, 随后在 btPID 里切片越界 panic; 读出错则表现为
  158. // "insert: load failed"。页留在缓存里, 并发读就一定命中最新内容, 不存在两头都不在的时刻。
  159. //
  160. // 也不再清空 dirtyCount: 快照里的页此刻【尚未落盘】, 计数必须继续按"未持久化"算。
  161. // 页的脏标记由提交方在 wal.Commit() 成功之后用 MarkClean 收掉 —— 只有快照后没被再改过
  162. // (ver 未变) 的页才会转成干净页, 改过的继续留脏, 等下一次提交。
  163. // 干净页照旧保留在池中继续命中 (旧实现无论脏净一律 delete, 等于每次提交都清空缓存,
  164. // 提交后的读取全部回落到磁盘); 容量仍由 evictCleanForRoomLocked 约束。
  165. func (bp *BufferPool) CollectDirty() []DirtyPage {
  166. // 锁定写入,防止新的写操作
  167. bp.LockWrites()
  168. defer bp.UnlockWrites()
  169. bp.mu.Lock()
  170. var dirty []DirtyPage
  171. for pid, entry := range bp.pages {
  172. if !entry.dirty {
  173. continue // 干净页: 与磁盘一致, 留在池中继续服务后续读
  174. }
  175. data := make([]byte, len(entry.data))
  176. copy(data, entry.data)
  177. dirty = append(dirty, DirtyPage{PageID: pid, Data: data, PageLen: entry.pageLen, ver: entry.ver})
  178. }
  179. // 这里没有释放任何页, 所以不唤醒等待容量的 goroutine (快照只是复制)。
  180. bp.mu.Unlock()
  181. return dirty
  182. }
  183. // MarkClean 把「确实已经落盘」的快照页转成干净页 (提交成功后由提交方调用)。
  184. //
  185. // 判据是版本号而不是"提交成功了": 提交窗口里写入线程可能已经把这一页又改了一遍
  186. // (缓存里是更新的内容, 而本次 WAL 记的是快照那一份)。这种页必须继续留脏 —— 否则它
  187. // 的新内容既不会被写盘, 脏标记还向引擎报告"已经持久化了", 该页一旦被淘汰就永久丢失。
  188. //
  189. // 快照里的页若在窗口里被别的路径丢弃 (回收 / 直写盘), 这里跳过即可: 那些路径自己
  190. // 负责缓存一致性。
  191. func (bp *BufferPool) MarkClean(pages []DirtyPage) {
  192. if len(pages) == 0 {
  193. return
  194. }
  195. bp.mu.Lock()
  196. defer bp.mu.Unlock()
  197. for _, p := range pages {
  198. entry, ok := bp.pages[p.PageID]
  199. if !ok || !entry.dirty || entry.ver != p.ver {
  200. continue
  201. }
  202. entry.dirty = false
  203. bp.dirtyCount--
  204. }
  205. if bp.dirtyCount < 0 {
  206. bp.dirtyCount = 0
  207. }
  208. }
  209. // stampLocked 给页条目盖一个新的版本号。调用方必须持有 bp.mu。
  210. //
  211. // 版本号只用于回答一个问题: "从我拍快照到现在, 这一页被改过吗"。
  212. func (bp *BufferPool) stampLocked(e *pageEntry) {
  213. bp.verSeq++
  214. e.ver = bp.verSeq
  215. }
  216. // RestoreDirty 提交失败时兜底: 保证这批页在缓存里仍然是脏的。
  217. //
  218. // CollectDirty 现在不再把页移出缓存, 所以正常路径上这里无事可做 —— 留着是因为错误
  219. // 处理期间别的动作 (页回收 / 直写盘) 也可能碰过这些页。只补【缺失】的条目: 页已经在
  220. // 缓存里就原样不动, 免得用一份旧快照盖掉提交窗口里写入线程刚改出来的新内容。
  221. //
  222. // 这里刻意不触发阈值回调, 避免在错误处理路径上递归触发 Commit。
  223. func (bp *BufferPool) RestoreDirty(pages []DirtyPage) {
  224. if len(pages) == 0 {
  225. return
  226. }
  227. bp.mu.Lock()
  228. defer bp.mu.Unlock()
  229. for _, p := range pages {
  230. if _, ok := bp.pages[p.PageID]; ok {
  231. continue // 已在缓存 (很可能还带着更新的内容), 不要覆盖
  232. }
  233. e := &pageEntry{data: p.Data, dirty: true, pageLen: p.PageLen}
  234. bp.stampLocked(e)
  235. bp.pages[p.PageID] = e
  236. bp.touchLocked(e)
  237. bp.dirtyCount++
  238. }
  239. }
  240. // DiscardPages 丢弃指定页的缓存 (不落盘), 并释放其占用容量。
  241. //
  242. // 事务 Rollback 使用: 事务期间 autoCommit 被 currentTxn 拦住, 任何脏页都不会刷到磁盘,
  243. // 所以「撤销」= 直接把本事务碰过的页从缓存里丢掉, 后续读取自然回到磁盘旧值。
  244. // 只丢弃传入的页 —— 事务开始前遗留的脏页不属于本事务, 必须原样保留。
  245. func (bp *BufferPool) DiscardPages(pageIDs map[uint32]struct{}) {
  246. if len(pageIDs) == 0 {
  247. return
  248. }
  249. bp.mu.Lock()
  250. defer bp.mu.Unlock()
  251. removedDirty := 0
  252. for pid := range pageIDs {
  253. entry, ok := bp.pages[pid]
  254. if !ok {
  255. continue
  256. }
  257. if entry.dirty {
  258. removedDirty++
  259. }
  260. bp.removeLocked(pid, entry)
  261. }
  262. bp.dirtyCount -= removedDirty
  263. if bp.dirtyCount < 0 {
  264. bp.dirtyCount = 0
  265. }
  266. }
  267. // DiscardPage 丢弃单页缓存 (不落盘), 并释放其占用容量。
  268. //
  269. // 用于「绕过缓冲池直写磁盘」(WritePageDirect) 之后, 让缓存中该页的旧副本立即失效,
  270. // 避免随后 ReadPage 命中与磁盘不一致的陈旧内容。
  271. func (bp *BufferPool) DiscardPage(pageID uint32) {
  272. bp.mu.Lock()
  273. entry, ok := bp.pages[pageID]
  274. if !ok {
  275. bp.mu.Unlock()
  276. return
  277. }
  278. if entry.dirty {
  279. bp.dirtyCount--
  280. if bp.dirtyCount < 0 {
  281. bp.dirtyCount = 0
  282. }
  283. }
  284. bp.removeLocked(pageID, entry)
  285. bp.mu.Unlock()
  286. }
  287. // Flush 将所有脏页写入磁盘 (Close / 维护操作时使用), 并把落盘成功的页转成干净页。
  288. //
  289. // 收尾必须走 MarkClean: CollectDirty 不再移出缓存, 少了这一步, 刚写完的页会一直带着
  290. // 脏标记, 下一次 Commit 会把同一批内容再写一遍 (并且写进 WAL)。
  291. // 写盘失败的页【保持 dirty】并计入返回值: 标干净等于宣布"已持久化", 该页一旦被淘汰
  292. // 就永久丢失 (静默丢数据)。返回第一个写盘错误, 全部成功返回 nil —— 调用方 (如
  293. // db.Close) 据此判断收尾是否可信; ver 比对仍会放过提交窗口里被改过的页, 它们继续
  294. // 留脏, 交给下一次 Flush / Commit。
  295. func (bp *BufferPool) Flush() error {
  296. dirty := bp.CollectDirty()
  297. var firstErr error
  298. var written []DirtyPage
  299. for _, item := range dirty {
  300. if err := bp.diskWrite(item.PageID, item.Data); err != nil {
  301. if firstErr == nil {
  302. firstErr = err
  303. }
  304. continue // 写失败的页保持 dirty, 等下一次 Flush / Commit 重试
  305. }
  306. written = append(written, item)
  307. }
  308. // 只把确实落盘的页标干净
  309. bp.MarkClean(written)
  310. return firstErr
  311. }
  312. // Close 关闭缓冲池 (刷盘后停止); 刷盘错误被丢弃 —— 需要错误判定请直接调 Flush
  313. func (bp *BufferPool) Close() {
  314. _ = bp.Flush()
  315. }
  316. // UpdatePage 更新缓存页 (WAL 提交后调用, 保持缓存与磁盘一致)
  317. func (bp *BufferPool) UpdatePage(pageID uint32, data []byte) {
  318. bp.mu.Lock()
  319. defer bp.mu.Unlock()
  320. cached := make([]byte, len(data))
  321. copy(cached, data)
  322. if old, ok := bp.pages[pageID]; ok {
  323. old.data = cached
  324. old.dirty = false
  325. old.pageLen = len(data)
  326. bp.stampLocked(old) // 内容换过了, 版本号必须往前走, 否则旧快照会误认为这页没变
  327. bp.touchLocked(old)
  328. } else {
  329. e := &pageEntry{data: cached, dirty: false, pageLen: len(data)}
  330. bp.stampLocked(e)
  331. bp.pages[pageID] = e
  332. bp.touchLocked(e)
  333. }
  334. // 回填的页是干净页, 不受 dirtyCount 约束; 不加这一步, 大量回收页会把缓存
  335. // 无限撑大 (回收 N 页 → 缓存增加 N 页), 并让 WritePage 的容量判断形同虚设。
  336. bp.evictCleanForRoomLocked()
  337. }
  338. // evictCleanForRoomLocked 容量已满时按时间戳 LRU 淘汰干净页, 直到降到容量以下。
  339. //
  340. // 采样淘汰 (对照 Redis 的 maxmemory-samples): Go map 的遍历顺序本身随机,
  341. // 从头取 evictSample 个候选、在其中选 lastTouch 最小的干净页 (最久未用) 淘汰。
  342. // 池子大于样本量时是 O(K) 近似 LRU; 小于等于样本量时覆盖全池, 就是精确 LRU。
  343. //
  344. // 为什么不做全表精确最小: 本函数在 WritePage 的持锁路径上, 池满后每次写入都触发,
  345. // 且一轮可能连续淘汰几十上百页 —— 全表扫是 O(n)/页, 采样把整轮压到 O(K×淘汰数)。
  346. // 4000 页池从每次 ~200-400μs 降到 ~3μs。读路径的 RLock + atomic touch 不受影响,
  347. // "一写多读"的并发收益原样保留。
  348. //
  349. // 脏页跳过 (计入样本数, 同 Redis) —— 淘汰脏页会静默丢掉尚未落盘的修改。
  350. // 若抽到的候选全是脏页, 本次不做任何事 —— 允许缓存临时超出容量, 由脏页阈值触发的
  351. // Commit 尽快回落; 下一次写入会重新随机抽样。
  352. //
  353. // 调用方必须持有 bp.mu (Lock).
  354. func (bp *BufferPool) evictCleanForRoomLocked() {
  355. for bp.capacity > 0 && len(bp.pages) >= bp.capacity {
  356. // 采样找 lastTouch 最小的干净页: Go map 遍历顺序随机, 取前
  357. // evictSample 个候选即随机样本, 池子 ≤ 样本量时退化为精确 LRU。
  358. var victimPID uint32
  359. var victim *pageEntry
  360. var minTouch int64 = -1
  361. n := 0
  362. for pid, e := range bp.pages {
  363. if n >= evictSample {
  364. break
  365. }
  366. n++
  367. if e.dirty {
  368. continue
  369. }
  370. if ts := atomic.LoadInt64(&e.lastTouch); victim == nil || ts < minTouch {
  371. minTouch, victimPID, victim = ts, pid, e
  372. }
  373. }
  374. if victim == nil {
  375. return // 样本内全是脏页, 允许临时超容量
  376. }
  377. bp.removeLocked(victimPID, victim)
  378. }
  379. }
  380. // ClearAll 清除所有脏标记, 保留页在缓存中。
  381. //
  382. // ⚠️ 这是"无差别"操作, 只适用于确定没有并发写入的维护场景 (关闭 / 建表删表等)。
  383. // 提交路径上【不能】用它: 提交窗口里被写入线程重新改脏的页并不在本笔 WAL 里, 一并
  384. // 标干净就等于宣布"已持久化", 那份新内容以后既不会被写盘, 也会被淘汰逻辑当作可以
  385. // 直接丢弃的干净页。提交成功要收脏标记请用 MarkClean (按版本号逐页判定)。
  386. func (bp *BufferPool) ClearAll() {
  387. bp.mu.Lock()
  388. defer bp.mu.Unlock()
  389. for _, entry := range bp.pages {
  390. entry.dirty = false
  391. }
  392. bp.dirtyCount = 0
  393. }
  394. // ReadPage 读页 — 先查缓存, 未命中则读磁盘并作为【干净页载入缓存】。
  395. //
  396. // ⚡ 并发读核心: cache-hit 路径完全用 RLock + atomic touch!
  397. // 多个 goroutine 同时 GetPage(不同页) 不再互相阻塞 —— 这是让 TsDB2
  398. // 真正支持"一写多读"的关键一步。
  399. //
  400. // 读缓存让重复扫描/点查不必每次都走系统调用; 载入的页与磁盘一致, 可被 LRU 淘汰。
  401. // 读盘期间另一线程可能已把该页写进缓存 (写缓存优先): 重新检查后以池内内容为准,
  402. // 绝不拿这份可能更旧的读盘结果覆盖脏页。直写盘路径 (WritePageDirect) 自带缓存失效,
  403. // 读缓存不会滞留陈旧内容。
  404. func (bp *BufferPool) ReadPage(pageID uint32, buf []byte) error {
  405. // cache-hit 路径: 用读锁查 map, 复制数据, atomic 打时间戳 —— 全程不需要写锁
  406. bp.mu.RLock()
  407. if entry, ok := bp.pages[pageID]; ok {
  408. copy(buf, entry.data)
  409. bp.touch(entry) // atomic Store, 不需要写锁
  410. bp.mu.RUnlock()
  411. return nil
  412. }
  413. bp.mu.RUnlock()
  414. // 缓存未命中 → 读磁盘 (锁外!)
  415. if err := bp.diskRead(pageID, buf); err != nil {
  416. return err
  417. }
  418. // 作为干净页入池。入池前 double-check: 读盘这段时间里别的线程可能已经写入了这一页
  419. // (缓存里是更新/未落盘的内容), 此时以池内为准, 丢弃读盘结果。
  420. bp.mu.Lock()
  421. defer bp.mu.Unlock()
  422. if entry, ok := bp.pages[pageID]; ok {
  423. copy(buf, entry.data)
  424. bp.touchLocked(entry)
  425. return nil
  426. }
  427. bp.evictCleanForRoomLocked()
  428. entry := &pageEntry{data: make([]byte, len(buf)), dirty: false, pageLen: len(buf)}
  429. copy(entry.data, buf)
  430. bp.pages[pageID] = entry
  431. bp.touchLocked(entry)
  432. return nil
  433. }
  434. // WritePage 写页 — 写入缓存并标记脏,不立即落盘
  435. func (bp *BufferPool) WritePage(pageID uint32, buf []byte) error {
  436. // 1. 等待写锁释放(Commit/Flush 期间阻塞)
  437. bp.waitForWriteLock()
  438. // 事务支持:通知 Transaction 缓存原始数据
  439. if bp.onPageWrite != nil {
  440. bp.onPageWrite(pageID)
  441. }
  442. var triggerThreshold bool
  443. bp.mu.Lock()
  444. // 2. 容量控制: 满了先按 LRU 淘汰干净页, 不阻塞等待。
  445. // 旧实现阻塞在 capacityWait 上等 Commit/Discard 释放空间, 但这两者都不可能
  446. // 在"当前这个正在阻塞的 goroutine"内被调用 —— 单写入者模型下必然死锁
  447. // (fatal error: all goroutines are asleep)。批量删除回收旧页时, realFreePage
  448. // 会把每个回收页经 SyncCache→UpdatePage 以干净页回填缓存(不计 dirtyCount),
  449. // 缓存被干净页迅速撑满, 紧接其后的写入即触发该死锁。
  450. bp.evictCleanForRoomLocked()
  451. existing, exists := bp.pages[pageID]
  452. switch {
  453. case exists && existing.dirty:
  454. // 已是脏页, 直接覆盖, 不增加脏页计数。
  455. //
  456. // 【原地覆写】旧实现在这里每次都新分配一块 len(buf) 的副本再整个换掉
  457. // existing.data —— 对逐行写入路径 (B+ 树叶 + 数据页 + 表信息页, 每行约 6 次
  458. // 16KB 写) 意味着每行 ~96KB 的纯垃圾, 5 万行灌数实测 ~4.7GB 分配, GC 压力
  459. // 吃掉约 1/3 的插入吞吐。这里改为: 页长一致时直接 copy 进旧缓冲 (旧缓冲没有
  460. // 任何外部别名 —— CollectDirty 的快照是独立副本, UpdatePage/RestoreDirty 各自
  461. // 持有自己的数据), 只留一次 16KB memcpy, 分配量从"每次写"降到"每张页"。
  462. if len(existing.data) == len(buf) {
  463. copy(existing.data, buf)
  464. } else {
  465. cached := make([]byte, len(buf))
  466. copy(cached, buf)
  467. existing.data = cached
  468. }
  469. bp.stampLocked(existing) // 内容变了: 让提交窗口里的旧快照认不出这一页
  470. bp.touchLocked(existing)
  471. case exists:
  472. // 干净页 (读缓存/回收回填的页) 被改写: 复用条目, 标脏
  473. if len(existing.data) == len(buf) {
  474. copy(existing.data, buf)
  475. } else {
  476. cached := make([]byte, len(buf))
  477. copy(cached, buf)
  478. existing.data = cached
  479. }
  480. existing.dirty = true
  481. existing.pageLen = len(buf)
  482. bp.stampLocked(existing)
  483. bp.dirtyCount++
  484. bp.touchLocked(existing)
  485. default:
  486. // 新页, 增加脏页计数
  487. cached := make([]byte, len(buf))
  488. copy(cached, buf)
  489. e := &pageEntry{data: cached, dirty: true, pageLen: len(buf)}
  490. bp.stampLocked(e)
  491. bp.pages[pageID] = e
  492. bp.touchLocked(e)
  493. bp.dirtyCount++
  494. }
  495. // 3. 检查是否超过阈值 —— 只按「脏页数」判定。
  496. //
  497. // 不能再用 len(bp.pages): CollectDirty 现在会保留干净页, 读缓存也会让干净页
  498. // 持续驻留, 总页数会持续偏高。若仍以总页数触发, 阈值会被这些干净页反复顶爆,
  499. // 提交频率凭空放大数倍。脏页数才是真正"尚未落盘"的数据量, 也正是崩溃时可能
  500. // 丢失的上限; 干净页的容量由 evictCleanForRoomLocked 按 capacity 约束。
  501. if bp.dirtyThreshold > 0 && bp.onThreshold != nil && bp.dirtyCount >= bp.dirtyThreshold {
  502. triggerThreshold = true
  503. }
  504. bp.mu.Unlock()
  505. // 4. 在锁外触发回调, 避免死锁
  506. if triggerThreshold {
  507. bp.onThreshold()
  508. }
  509. return nil
  510. }
  511. // Len 当前缓存页数 (调试用)
  512. func (bp *BufferPool) Len() int {
  513. bp.mu.Lock()
  514. defer bp.mu.Unlock()
  515. return len(bp.pages)
  516. }
  517. // DirtyCount 脏页数 (调试用)
  518. func (bp *BufferPool) DirtyCount() int {
  519. bp.mu.Lock()
  520. defer bp.mu.Unlock()
  521. return bp.dirtyCount
  522. }
  523. // SetCapacity 设置 BufferPool 容量(页数)
  524. func (bp *BufferPool) SetCapacity(capacity int) {
  525. bp.mu.Lock()
  526. bp.capacity = capacity
  527. bp.mu.Unlock()
  528. }
  529. // SetOnPageWriteCallback 设置页写入回调(用于事务缓存原始数据)
  530. func (bp *BufferPool) SetOnPageWriteCallback(callback func(uint32)) {
  531. bp.mu.Lock()
  532. bp.onPageWrite = callback
  533. bp.mu.Unlock()
  534. }