transaction.go 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. package core
  2. import (
  3. "fmt"
  4. )
  5. // Transaction 事务管理器
  6. //
  7. // 核心思想:
  8. // - 事务开始时,记录所有被修改页的原始数据(快照)
  9. // - 事务内操作正常修改 BufferPool
  10. // - Commit 时将修改后的脏页写盘(WAL + fsync)
  11. // - Rollback 时用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
  12. type Transaction struct {
  13. db *DB
  14. originalPages map[uint32][]byte // pageID → 原始数据快照(用于 Rollback)
  15. committed bool
  16. rolledBack bool
  17. }
  18. // BeginTransaction 开启新事务
  19. func (db *DB) BeginTransaction() *Transaction {
  20. txn := &Transaction{
  21. db: db,
  22. originalPages: make(map[uint32][]byte),
  23. committed: false,
  24. rolledBack: false,
  25. }
  26. // 设置当前活跃事务
  27. db.currentTxn = txn
  28. return txn
  29. }
  30. // cacheOriginalPage 缓存页的原始数据(首次修改时调用)
  31. func (txn *Transaction) cacheOriginalPage(pageID uint32) error {
  32. // 如果已经缓存过,不需要重复缓存
  33. if _, exists := txn.originalPages[pageID]; exists {
  34. return nil
  35. }
  36. // 从磁盘读取原始数据
  37. buf := make([]byte, PageSize)
  38. if err := txn.db.file.readPageDisk(pageID, buf); err != nil {
  39. return fmt.Errorf("failed to read page %d for rollback cache: %v", pageID, err)
  40. }
  41. txn.originalPages[pageID] = buf
  42. return nil
  43. }
  44. // Commit 提交事务
  45. // 1. 收集所有 BufferPool 脏页
  46. // 2. 写入 WAL
  47. // 3. ⭐ 关键:fsync 确保落盘 ⭐
  48. // 4. 回收 WAL 页
  49. // 5. 清理原始数据快照(释放内存)
  50. func (txn *Transaction) Commit() error {
  51. if txn.committed {
  52. return fmt.Errorf("transaction already committed")
  53. }
  54. if txn.rolledBack {
  55. return fmt.Errorf("transaction already rolled back")
  56. }
  57. // 1. 收集所有脏页
  58. dirty := txn.db.file.CollectDirty()
  59. if len(dirty) == 0 {
  60. txn.committed = true
  61. txn.db.currentTxn = nil
  62. txn.originalPages = nil // ← 清理原始数据快照
  63. return nil // 无脏页,直接返回
  64. }
  65. // 2. 创建 WAL 事务
  66. wal := txn.db.BeginTxn()
  67. // 3. 记录所有脏页到 WAL
  68. for _, d := range dirty {
  69. wal.RecordPage(d.PageID, d.Data)
  70. }
  71. // 4. ⭐ 关键:Commit 会触发 fsync,确保落盘 ⭐
  72. if err := wal.Commit(); err != nil {
  73. return fmt.Errorf("WAL commit failed: %v", err)
  74. }
  75. // 5. 清理事务状态 + 释放原始数据快照
  76. txn.committed = true
  77. txn.db.currentTxn = nil
  78. txn.originalPages = nil // ← 提交成功,原始数据不再需要
  79. return nil
  80. }
  81. // Rollback 回滚事务
  82. // 用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
  83. // 回滚后清理所有缓存数据(释放内存)
  84. func (txn *Transaction) Rollback() error {
  85. if txn.committed {
  86. return fmt.Errorf("transaction already committed, cannot rollback")
  87. }
  88. if txn.rolledBack {
  89. return fmt.Errorf("transaction already rolled back")
  90. }
  91. // 用原始数据覆盖所有被修改的页
  92. for pageID, originalData := range txn.originalPages {
  93. // 直接写回 BufferPool(绕过 WAL)
  94. if err := txn.db.file.WritePage(pageID, originalData); err != nil {
  95. return fmt.Errorf("failed to rollback page %d: %v", pageID, err)
  96. }
  97. }
  98. // 清理事务状态 + 释放所有缓存数据
  99. txn.rolledBack = true
  100. txn.db.currentTxn = nil
  101. txn.originalPages = nil // ← 回滚成功,原始数据和修改数据都不再需要
  102. return nil
  103. }
  104. // IsCommitted 检查事务是否已提交
  105. func (txn *Transaction) IsCommitted() bool {
  106. return txn.committed
  107. }
  108. // IsRolledBack 检查事务是否已回滚
  109. func (txn *Transaction) IsRolledBack() bool {
  110. return txn.rolledBack
  111. }