| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132 |
- package core
- import (
- "fmt"
- )
- // Transaction 事务管理器
- //
- // 核心思想:
- // - 事务开始时,记录所有被修改页的原始数据(快照)
- // - 事务内操作正常修改 BufferPool
- // - Commit 时将修改后的脏页写盘(WAL + fsync)
- // - Rollback 时用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
- type Transaction struct {
- db *DB
- originalPages map[uint32][]byte // pageID → 原始数据快照(用于 Rollback)
- committed bool
- rolledBack bool
- }
- // BeginTransaction 开启新事务
- func (db *DB) BeginTransaction() *Transaction {
- txn := &Transaction{
- db: db,
- originalPages: make(map[uint32][]byte),
- committed: false,
- rolledBack: false,
- }
- // 设置当前活跃事务
- db.currentTxn = txn
- return txn
- }
- // cacheOriginalPage 缓存页的原始数据(首次修改时调用)
- func (txn *Transaction) cacheOriginalPage(pageID uint32) error {
- // 如果已经缓存过,不需要重复缓存
- if _, exists := txn.originalPages[pageID]; exists {
- return nil
- }
- // 从磁盘读取原始数据
- buf := make([]byte, PageSize)
- if err := txn.db.file.readPageDisk(pageID, buf); err != nil {
- return fmt.Errorf("failed to read page %d for rollback cache: %v", pageID, err)
- }
- txn.originalPages[pageID] = buf
- return nil
- }
- // Commit 提交事务
- // 1. 收集所有 BufferPool 脏页
- // 2. 写入 WAL
- // 3. ⭐ 关键:fsync 确保落盘 ⭐
- // 4. 回收 WAL 页
- // 5. 清理原始数据快照(释放内存)
- func (txn *Transaction) Commit() error {
- if txn.committed {
- return fmt.Errorf("transaction already committed")
- }
- if txn.rolledBack {
- return fmt.Errorf("transaction already rolled back")
- }
- // 1. 收集所有脏页
- dirty := txn.db.file.CollectDirty()
- if len(dirty) == 0 {
- txn.committed = true
- txn.db.currentTxn = nil
- txn.originalPages = nil // ← 清理原始数据快照
- return nil // 无脏页,直接返回
- }
- // 2. 创建 WAL 事务
- wal := txn.db.BeginTxn()
- // 3. 记录所有脏页到 WAL
- for _, d := range dirty {
- wal.RecordPage(d.PageID, d.Data)
- }
- // 4. ⭐ 关键:Commit 会触发 fsync,确保落盘 ⭐
- if err := wal.Commit(); err != nil {
- return fmt.Errorf("WAL commit failed: %v", err)
- }
- // 5. 清理事务状态 + 释放原始数据快照
- txn.committed = true
- txn.db.currentTxn = nil
- txn.originalPages = nil // ← 提交成功,原始数据不再需要
- return nil
- }
- // Rollback 回滚事务
- // 用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
- // 回滚后清理所有缓存数据(释放内存)
- func (txn *Transaction) Rollback() error {
- if txn.committed {
- return fmt.Errorf("transaction already committed, cannot rollback")
- }
- if txn.rolledBack {
- return fmt.Errorf("transaction already rolled back")
- }
- // 用原始数据覆盖所有被修改的页
- for pageID, originalData := range txn.originalPages {
- // 直接写回 BufferPool(绕过 WAL)
- if err := txn.db.file.WritePage(pageID, originalData); err != nil {
- return fmt.Errorf("failed to rollback page %d: %v", pageID, err)
- }
- }
- // 清理事务状态 + 释放所有缓存数据
- txn.rolledBack = true
- txn.db.currentTxn = nil
- txn.originalPages = nil // ← 回滚成功,原始数据和修改数据都不再需要
- return nil
- }
- // IsCommitted 检查事务是否已提交
- func (txn *Transaction) IsCommitted() bool {
- return txn.committed
- }
- // IsRolledBack 检查事务是否已回滚
- func (txn *Transaction) IsRolledBack() bool {
- return txn.rolledBack
- }
|