package core import ( "TsDB2/DB/core/pool" "os" "unsafe" ) // File 只负责:打开、关闭、读、写 —— unsafe 全部收敛 type File struct { f *os.File bp *pool.BufferPool // 页面缓冲池 (nil = 未启用) } func OpenFile(path string) (*File, error) { f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0644) if err != nil { return nil, err } return &File{f: f}, nil } func (ff *File) Close() error { if ff.bp != nil { ff.bp.Close() // 先刷盘再关文件 } return ff.f.Close() } // InitBufferPool 初始化页面缓冲池, 拦截 ReadPage/WritePage 实现内存缓存 // capacity: 最大缓存页数 (建议 10000~50000) func (ff *File) InitBufferPool(capacity int) { ff.bp = pool.NewBufferPool(capacity, ff.readPageDisk, ff.writePageDisk) } // SetDirtyThreshold 设置脏页阈值, 超过时触发回调 (配合 WAL 自动 Commit) func (ff *File) SetDirtyThreshold(threshold int, onThreshold func()) { if ff.bp != nil { ff.bp.SetDirtyThreshold(threshold, onThreshold) } } // CollectDirty 收集所有脏页快照并清除脏标记 (配合 WAL 使用) func (ff *File) CollectDirty() []pool.DirtyPage { if ff.bp != nil { return ff.bp.CollectDirty() } return nil } // FlushBufferPool 刷盘 (仅在 Close 时使用) func (ff *File) FlushBufferPool() { if ff.bp != nil { ff.bp.Flush() } } func (ff *File) WritePage16K(pageID uint32, buf *[PageSize]byte) error { // 头占一个4K,所以 pageID 从 1 开始 offset := int64(PageSize4K) + int64(pageID-1)*int64(PageSize) _, err := ff.f.WriteAt(buf[:], offset) return err } // ----------------------------------------------------------------------------------- // ✅ 动态长度:索引页 — 缓冲池拦截层 // ----------------------------------------------------------------------------------- func (ff *File) WritePage(pageID uint32, buf []byte) error { if ff.bp != nil { return ff.bp.WritePage(pageID, buf) // 写缓存, 延迟落盘 } return ff.writePageDisk(pageID, buf) } func (ff *File) ReadPage(pageID uint32, buf []byte) error { if ff.bp != nil { return ff.bp.ReadPage(pageID, buf) // 先查缓存 } return ff.readPageDisk(pageID, buf) } // WritePageDirect 直接写盘 (绕过 BufferPool, 用于 WAL 回写) func (ff *File) WritePageDirect(pageID uint32, buf []byte) error { return ff.writePageDisk(pageID, buf) } // SyncCache 同步缓存页 (WAL 提交后调用, 保持缓存与磁盘一致) func (ff *File) SyncCache(pageID uint32, data []byte) { if ff.bp != nil { ff.bp.UpdatePage(pageID, data) } } // ClearCache 清空所有缓存 (WAL 提交后调用) func (ff *File) ClearCache() { if ff.bp != nil { ff.bp.ClearAll() } } // ----------------------------------------------------------------------------------- // ✅ 底层磁盘 I/O (缓冲池回源) // ----------------------------------------------------------------------------------- func (ff *File) writePageDisk(pageID uint32, buf []byte) error { offset := int64(PageSize4K) + int64(pageID-1)*int64(len(buf)) _, err := ff.f.WriteAt(buf, offset) return err } func (ff *File) readPageDisk(pageID uint32, buf []byte) error { offset := int64(PageSize4K) + int64(pageID-1)*int64(len(buf)) _, err := ff.f.ReadAt(buf, offset) return err } // ----------------------------------------------------------------------------------- // ✅ 头读写(固定4K) // ----------------------------------------------------------------------------------- func (ff *File) WriteHeader(h *HeaderPage) error { var buf [PageSize4K]byte copy(buf[:], (*[PageSize4K]byte)(unsafe.Pointer(h))[:]) _, err := ff.f.WriteAt(buf[:], 0) return err } func (ff *File) ReadHeader() (*HeaderPage, error) { buf := make([]byte, PageSize4K) _, err := ff.f.ReadAt(buf, 0) if err != nil { return nil, err } h := &HeaderPage{} copy((*[PageSize4K]byte)(unsafe.Pointer(h))[:], buf) return h, nil } func (ff *File) TruncateSize(size int64) error { return ff.f.Truncate(size) }