|
|
@@ -1,6 +1,7 @@
|
|
|
package pool
|
|
|
|
|
|
import (
|
|
|
+ "errors"
|
|
|
"sync"
|
|
|
)
|
|
|
|
|
|
@@ -32,6 +33,10 @@ type BufferPool struct {
|
|
|
dirtyCount int // 当前脏页数 (实时追踪)
|
|
|
diskRead func(pageID uint32, buf []byte) error // 底层磁盘读
|
|
|
diskWrite func(pageID uint32, buf []byte) error // 底层磁盘写
|
|
|
+
|
|
|
+ // 锁表机制
|
|
|
+ writeLocked bool // 写锁标志
|
|
|
+ writeChan chan struct{} // 用于通知解锁的 channel
|
|
|
}
|
|
|
|
|
|
// NewBufferPool 创建页面缓冲池
|
|
|
@@ -59,12 +64,50 @@ func (bp *BufferPool) SetDirtyThreshold(threshold int, onThreshold func()) {
|
|
|
bp.onThreshold = onThreshold
|
|
|
}
|
|
|
|
|
|
+// LockWrites 锁定写入(Commit/Flush 期间调用)
|
|
|
+func (bp *BufferPool) LockWrites() {
|
|
|
+ bp.mu.Lock()
|
|
|
+ bp.writeLocked = true
|
|
|
+ bp.writeChan = make(chan struct{}) // 创建新的 channel
|
|
|
+ bp.mu.Unlock()
|
|
|
+}
|
|
|
+
|
|
|
+// UnlockWrites 解锁写入(Commit/Flush 完成后调用)
|
|
|
+func (bp *BufferPool) UnlockWrites() {
|
|
|
+ bp.mu.Lock()
|
|
|
+ bp.writeLocked = false
|
|
|
+ if bp.writeChan != nil {
|
|
|
+ close(bp.writeChan) // 通知所有等待的 goroutine
|
|
|
+ bp.writeChan = nil
|
|
|
+ }
|
|
|
+ bp.mu.Unlock()
|
|
|
+}
|
|
|
+
|
|
|
+// waitForWriteLock 等待写锁释放(在 WritePage 中调用)
|
|
|
+func (bp *BufferPool) waitForWriteLock() {
|
|
|
+ bp.mu.Lock()
|
|
|
+ if !bp.writeLocked {
|
|
|
+ bp.mu.Unlock()
|
|
|
+ return // 未锁定,直接返回
|
|
|
+ }
|
|
|
+
|
|
|
+ // 保存 channel 引用
|
|
|
+ ch := bp.writeChan
|
|
|
+ bp.mu.Unlock()
|
|
|
+
|
|
|
+ // 阻塞等待解锁(channel 关闭时会返回)
|
|
|
+ <-ch
|
|
|
+}
|
|
|
+
|
|
|
// CollectDirty 取出所有脏页快照并清除脏标记
|
|
|
// 由 DB.Commit() 调用,配合 WAL 流程统一写回
|
|
|
// 同时清除干净页面缓存,释放内存并确保后续脏页计数准确
|
|
|
func (bp *BufferPool) CollectDirty() []DirtyPage {
|
|
|
+ // 1. 锁定写入,防止新的写操作
|
|
|
+ bp.LockWrites()
|
|
|
+ defer bp.UnlockWrites()
|
|
|
+
|
|
|
bp.mu.Lock()
|
|
|
- defer bp.mu.Unlock()
|
|
|
var dirty []DirtyPage
|
|
|
for pid, entry := range bp.pages {
|
|
|
if entry.dirty {
|
|
|
@@ -78,6 +121,7 @@ func (bp *BufferPool) CollectDirty() []DirtyPage {
|
|
|
delete(bp.pages, pid)
|
|
|
}
|
|
|
bp.dirtyCount = 0 // 重置脏页计数
|
|
|
+ bp.mu.Unlock()
|
|
|
return dirty
|
|
|
}
|
|
|
|
|
|
@@ -130,11 +174,22 @@ func (bp *BufferPool) ReadPage(pageID uint32, buf []byte) error {
|
|
|
|
|
|
// WritePage 写页 — 写入缓存并标记脏,不立即落盘
|
|
|
func (bp *BufferPool) WritePage(pageID uint32, buf []byte) error {
|
|
|
+ // 1. 等待写锁释放(Commit/Flush 期间阻塞)
|
|
|
+ bp.waitForWriteLock()
|
|
|
+
|
|
|
cached := make([]byte, len(buf))
|
|
|
copy(cached, buf)
|
|
|
|
|
|
var triggerThreshold bool
|
|
|
+ var err error
|
|
|
bp.mu.Lock()
|
|
|
+
|
|
|
+ // 2. 检查 BufferPool 容量
|
|
|
+ if bp.capacity > 0 && len(bp.pages) >= bp.capacity {
|
|
|
+ bp.mu.Unlock()
|
|
|
+ return errors.New("buffer pool is full, please commit first")
|
|
|
+ }
|
|
|
+
|
|
|
existing, exists := bp.pages[pageID]
|
|
|
if exists && existing.dirty {
|
|
|
// 已是脏页, 直接覆盖, 不增加脏页计数
|
|
|
@@ -145,23 +200,21 @@ func (bp *BufferPool) WritePage(pageID uint32, buf []byte) error {
|
|
|
bp.dirtyCount++
|
|
|
}
|
|
|
|
|
|
- // 检查是否超过阈值 (脏页或总页面数)
|
|
|
- // 使用总页面数作为第二道安全网,防止脏页计数增长缓慢时内存无限膨胀
|
|
|
+ // 3. 检查是否超过阈值 (脏页或总页面数)
|
|
|
totalPages := len(bp.pages)
|
|
|
if bp.dirtyThreshold > 0 && bp.onThreshold != nil {
|
|
|
- // 脏页阈值或总页面数阈值 (两者相同,确保及时刷盘)
|
|
|
if bp.dirtyCount >= bp.dirtyThreshold || totalPages >= bp.dirtyThreshold {
|
|
|
triggerThreshold = true
|
|
|
}
|
|
|
}
|
|
|
bp.mu.Unlock()
|
|
|
|
|
|
- // 在锁外触发回调, 避免死锁
|
|
|
+ // 4. 在锁外触发回调, 避免死锁
|
|
|
if triggerThreshold {
|
|
|
bp.onThreshold()
|
|
|
}
|
|
|
|
|
|
- return nil
|
|
|
+ return err
|
|
|
}
|
|
|
|
|
|
// Len 当前缓存页数 (调试用)
|