transaction.go 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194
  1. package core
  2. import (
  3. "fmt"
  4. )
  5. // Transaction 事务管理器
  6. //
  7. // 核心思想:
  8. // - 事务内所有修改只落在 BufferPool —— autoCommit 被 currentTxn 拦住, 中途不会刷盘,
  9. // 所以磁盘始终是事务开始前的状态, 不需要任何「原始数据快照」
  10. // - Commit 时把脏页 + Header 整体写入 WAL, 保证原子
  11. // - Rollback 只需丢弃本事务碰过的页缓存(并从磁盘重载 TableInfo), 后续读取自然回到旧值
  12. type Transaction struct {
  13. db *DB
  14. touched map[uint32]struct{} // 本事务修改过的页 ID(Rollback 时丢弃其缓存)
  15. // pendingFree 本事务登记待回收的页 (堆页摘链 / 批量删除后作废的旧 DataPage)。
  16. // 不在事务中途直写盘: 提交成功后才真正还链, 回滚时整体丢弃
  17. // (回滚后链指针回到事务前, 这些页仍被引用, 绝不能还链)。
  18. pendingFree []uint32
  19. // pendingFreeChain 本事务登记待回收的整链 (如哈希索引整体重建后的旧桶链)。
  20. // 与 pendingFree 同一时序约定, 但一条链只登记一次, 提交后一次 FreeChain 还链。
  21. pendingFreeChain []pendingChain
  22. // allocated 本事务新分配的页 ID (从空闲链摘出, 或空闲链耗尽时由 NextFreePage 递增)。
  23. // 分配在 newPageWithReader 里就地推进了 header 的游离指针并直写盘, 事务中途无法回退;
  24. // 因此 Rollback 必须把这些页还给空闲链 —— 否则它们既不被任何表/索引引用、也不在
  25. // 空闲链上, 就成了永久泄漏的孤儿页 (文件只涨不缩)。
  26. // Commit 成功时这些页已被数据引用, 直接丢弃本表即可。
  27. allocated []uint32
  28. committed bool
  29. rolledBack bool
  30. }
  31. // 事务内被直接 FreePage(立即还链) 的页, 从 allocated 中撤销登记, 避免 Rollback 二次归还。
  32. func (txn *Transaction) undoAllocated(pageID uint32) {
  33. for i, id := range txn.allocated {
  34. if id == pageID {
  35. txn.allocated = append(txn.allocated[:i], txn.allocated[i+1:]...)
  36. return
  37. }
  38. }
  39. }
  40. // BeginTransaction 开启新事务
  41. func (db *DB) BeginTransaction() *Transaction {
  42. if db.errIfReadOnly() != nil {
  43. panic(ErrReadOnly) // 事务无 error 返回, 抢救模式下显式拒绝
  44. }
  45. txn := &Transaction{
  46. db: db,
  47. touched: make(map[uint32]struct{}),
  48. committed: false,
  49. rolledBack: false,
  50. }
  51. // 设置当前活跃事务
  52. db.currentTxn.Store(txn)
  53. return txn
  54. }
  55. // markTouched 登记本事务修改过的页 (由 BufferPool 的 onPageWrite 回调触发)。
  56. // 只记页 ID(4B/页), 不存页数据 —— 回滚靠「丢弃缓存 + 从磁盘重读」实现。
  57. func (txn *Transaction) markTouched(pageID uint32) {
  58. txn.touched[pageID] = struct{}{}
  59. }
  60. // Commit 提交事务
  61. // 只写「整页镜像」条目: 恢复所有被修改页的结构与数据
  62. // (B+树节点, IndexMeta, HeaderPage, DataPage 一次到位)。
  63. //
  64. // 不再额外写行级条目: 整页镜像已完整包含行数据, 再叠加行级记录会在恢复时被
  65. // 重复应用 (INSERT 变成两行), 见 wal_recovery 的 Phase 1 / Phase 2。
  66. func (txn *Transaction) Commit() error {
  67. if txn.committed {
  68. return fmt.Errorf("transaction already committed")
  69. }
  70. if txn.rolledBack {
  71. return fmt.Errorf("transaction already rolled back")
  72. }
  73. // 与 DB.Commit 共用同一把提交锁: 事务提交和自动提交(定时器 / 脏页阈值)也可能撞
  74. // 在一起, 两笔提交同时在飞会把页的归属搞乱。详见 DB.Commit 的说明。
  75. txn.db.commitMu.Lock()
  76. defer txn.db.commitMu.Unlock()
  77. // 1. 收集 BufferPool 脏页
  78. dirty := txn.db.file.CollectDirty()
  79. if len(dirty) == 0 {
  80. // 无脏页 = 修改已持久, 本事务登记的待回收页 + 全局积压都可以安全还链
  81. txn.db.freeNow(txn.pendingFree)
  82. txn.db.freeChainNow(txn.pendingFreeChain)
  83. txn.db.drainPendingFree()
  84. txn.committed = true
  85. txn.db.currentTxn.Store(nil)
  86. txn.touched = nil
  87. txn.pendingFree = nil
  88. txn.pendingFreeChain = nil
  89. txn.allocated = nil
  90. return nil
  91. }
  92. // 2. 创建 WAL 事务
  93. wal := txn.db.BeginTxn()
  94. // 3. 写整页条目: 所有被修改的页 (保证页结构完整恢复)
  95. // 数据源就是 BufferPool 脏页本身(最新数据), 事务期间没有中途刷盘,
  96. // 因此不存在"写回事务前旧快照覆盖新数据"的风险。
  97. for _, d := range dirty {
  98. wal.RecordPage(d.PageID, d.Data)
  99. }
  100. // 4. HeaderPage 也纳入
  101. // marshalHeader 读整张头页, 含提交路径正在改的 WAL 链指针, 必须在 headerMu 内拍快照。
  102. txn.db.file.withHeader(func() {
  103. wal.RecordPage(0, marshalHeader(txn.db.header))
  104. })
  105. // 5. ⭐ 关键:Commit 会触发 fsync,确保落盘 ⭐
  106. if err := wal.Commit(); err != nil {
  107. // 提交失败: 这批页继续留在缓存里当脏页(CollectDirty 不再移出缓存),
  108. // RestoreDirty 只做兜底, 见 DB.Commit 里的说明
  109. txn.db.file.RestoreDirty(dirty)
  110. return fmt.Errorf("WAL commit failed: %v", err)
  111. }
  112. // 6. 提交成功 → 把这一批页收成干净页 (按快照版本号, 提交窗口里被改过的页继续留脏)
  113. txn.db.file.MarkClean(dirty)
  114. // 7. 提交成功 → 堆链/索引指针已持久化, 本事务登记摘链的页现在可以真正还链。
  115. // 必须在 wal.Commit() 之后: 它内部已完成 fsync + 回写真实页。
  116. txn.db.freeNow(txn.pendingFree)
  117. txn.db.freeChainNow(txn.pendingFreeChain)
  118. txn.db.drainPendingFree()
  119. // 8. 清理事务状态
  120. txn.committed = true
  121. txn.db.currentTxn.Store(nil)
  122. txn.touched = nil
  123. txn.pendingFree = nil
  124. txn.pendingFreeChain = nil
  125. txn.allocated = nil
  126. return nil
  127. }
  128. // Rollback 回滚事务: 丢弃本事务修改过的页缓存, 从磁盘重新加载。
  129. //
  130. // 事务期间 autoCommit 被 currentTxn 拦住, 没有任何脏页被刷到磁盘, 磁盘上仍是事务
  131. // 开始前的状态 —— 所以"丢掉缓存"就等价于"撤销内存修改", 不需要任何页级快照。
  132. // 只丢 txn.touched 里的页: 事务开始前遗留的脏页不属于本事务, 必须原样保留。
  133. func (txn *Transaction) Rollback() error {
  134. if txn.committed {
  135. return fmt.Errorf("transaction already committed, cannot rollback")
  136. }
  137. if txn.rolledBack {
  138. return fmt.Errorf("transaction already rolled back")
  139. }
  140. // 1. 丢弃本事务碰过的页缓存 (不落盘), 后续读取自然回落到磁盘旧值
  141. txn.db.file.DiscardPages(txn.touched)
  142. // 2. Table.info 是独立于页缓存的内存副本, 必须从磁盘重新读回,
  143. // 否则 TotalRows / FirstData 等仍停留在事务中的值
  144. txn.db.tables.ReloadInfos(txn.touched)
  145. // 3. 本事务登记待回收的页/链必须整体丢弃: 回滚后堆链指针回到事务前,
  146. // 这些页重新被引用, 绝不能还链 (全局 pendingFree 不属于本事务, 保持不动)
  147. txn.pendingFree = nil
  148. txn.pendingFreeChain = nil
  149. // 4. 归还本事务新分配的页 (见 allocated 字段注释): 分配时 header 的游离指针
  150. // 已被就地推进并直写盘, 回滚不补回这些页就会产生孤儿页。
  151. // 放在 DiscardPages 之后: 池里那些未落盘的脏内容已丢弃, 这里用干净的空闲页头覆盖。
  152. txn.db.freeNow(txn.allocated)
  153. txn.allocated = nil
  154. // 清理事务状态
  155. txn.rolledBack = true
  156. txn.db.currentTxn.Store(nil)
  157. txn.touched = nil
  158. return nil
  159. }
  160. // IsCommitted 检查事务是否已提交
  161. func (txn *Transaction) IsCommitted() bool {
  162. return txn.committed
  163. }
  164. // IsRolledBack 检查事务是否已回滚
  165. func (txn *Transaction) IsRolledBack() bool {
  166. return txn.rolledBack
  167. }