package core import ( "git.ycl.pub/yl32940/TsDB2/DB/core/page" "git.ycl.pub/yl32940/TsDB2/DB/core/pool" "os" "sync" "unsafe" ) // File 只负责:打开、关闭、读、写 —— unsafe 全部收敛 type File struct { f *os.File bp *pool.BufferPool // 页面缓冲池 (nil = 未启用) // headerMu 保护 HeaderPage (pageID=0) 的【内存字段】及其持久化。 // // 头页是整个引擎唯一的全局可变结构: 写入路径要它分配 / 回收页号 (NextFreePage / // FreePageStart / FreePageEnd / FreePageCount), 提交路径要它发布 / 清空 WAL 链指针 // (FirstWALPage / LastWALPage) 并推进 NextTxnID。两条路径并发时若不串行化: // - 两个 goroutine 同时读到同一个 NextFreePage 再各自 +1 → 同一个页号发给两方, // 后写的一方覆盖先写的一方 → 静默丢数据; // - WriteHeader 在拷贝整张头页时, 另一个 goroutine 正在改字段 → 数据竞争 (崩溃)。 // 所有对 HeaderPage 字段的读写都必须落在 headerMu 内。 // // 不能把 mutex 放进 HeaderPage 本身: 它按内存布局经 unsafe 直接落盘 (见 WriteHeader / // marshalHeader), 加字段会改变盘上格式。 headerMu sync.Mutex // headerDirty 头页脏标记 (headerMu 保护): 自上次头页落盘以来 allocPage 是否 // 又改过头页 (NextFreePage 游标 / 空闲链)。 // // 旧实现在 allocPage 里每分配【一个】页就同步 pwrite 一次 4K 头页 —— 批量灌数 // 期间上千张新页 (数据页 / 树节点 / 堆页) 就是上千次头页落盘, 实测占灌数 CPU // 的 ~16%。而这期间数据页本身全在缓冲池里 (未 Commit 根本不落盘), 头页急着 // 持久化毫无收益 —— 崩溃后这批缓冲页本来就丢了, 头页游标回退一点反而自洽 // (那些页号还没被任何盘上数据占用)。 // // 落盘点收敛到三处: // - Commit: WAL 打包写头页 (wal.go 的 writeHeaderLocked 链指针收尾); // - 空闲链变更: realFreePage / recycleChain 本来就写头页; // - Close: db.Close 先 Commit (写头页) 再刷缓冲池, 兜底用 FlushHeader。 // 前两处之外若有人要把数据页直接写盘, 必须先让头页落盘 (页号 > 盘上 // NextFreePage 的页一旦落盘, 重开时会被二次分配覆盖)。 headerDirty bool } 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) } } // SetOnPageWriteCallback 设置页写入回调(用于事务登记被修改的页) func (ff *File) SetOnPageWriteCallback(callback func(pageID uint32)) { if ff.bp != nil { ff.bp.SetOnPageWriteCallback(callback) } } // LockWrites 锁定写入 —— 锁的是"页落盘"(BufferPool.WritePage), 不是写入动作本身。 func (ff *File) LockWrites() { if ff.bp != nil { ff.bp.LockWrites() } } // UnlockWrites 解锁写入 (见 LockWrites 的语义说明) func (ff *File) UnlockWrites() { if ff.bp != nil { ff.bp.UnlockWrites() } } // SetBufferPoolCapacity 设置 BufferPool 容量(页数) func (ff *File) SetBufferPoolCapacity(capacity int) { if ff.bp != nil { ff.bp.SetCapacity(capacity) } } // CollectDirty 收集所有脏页快照 (配合 WAL 使用; 页不会因此离开缓存, 见 pool.CollectDirty) func (ff *File) CollectDirty() []pool.DirtyPage { if ff.bp != nil { return ff.bp.CollectDirty() } return nil } // MarkClean 把这批快照页收成干净页 (提交成功后调用, 与 CollectDirty 配对) func (ff *File) MarkClean(pages []pool.DirtyPage) { if ff.bp != nil { ff.bp.MarkClean(pages) } } // FlushBufferPool 刷盘 (仅在 Close 时使用) func (ff *File) FlushBufferPool() { if ff.bp != nil { ff.bp.Flush() } } // RestoreDirty 提交失败时兜底: 保证这批页在缓存里仍然是脏页 (与 CollectDirty 配对) func (ff *File) RestoreDirty(pages []pool.DirtyPage) { if ff.bp != nil { ff.bp.RestoreDirty(pages) } } // DiscardPages 丢弃指定页的缓存 (不落盘), 供事务 Rollback 回到磁盘状态 func (ff *File) DiscardPages(pageIDs map[uint32]struct{}) { if ff.bp != nil { ff.bp.DiscardPages(pageIDs) } } 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) // // 因为绕过了缓冲池, 必须让缓存中该页的旧副本失效 —— 否则后续 ReadPage 会命中 // 与磁盘不一致的陈旧内容。典型场景: 页从空闲链复用时缓存里还留有旧副本, 被当作 // WAL 页直写后该副本已失真; 整链回收若只同步链尾页, 就会漏掉链内这些页。 // // 失效必须排在【落盘之后】。反过来的话, 从删缓存到写完盘这一小段里, 这一页同样 // "两头都不在": 对一页本会话刚分配、盘上还从没写过的页, 并发读会拿到全零 (文件还 // 没长到那个偏移时直接读出错), 而 btree 把全零页当成 EntryLen==0 的节点, 随后就在 // btPID 里切片越界。先落盘再失效, 最坏也只是读到"马上要被覆盖的旧内容", 不会两头皆空。 func (ff *File) WritePageDirect(pageID uint32, buf []byte) error { if err := ff.writePageDisk(pageID, buf); err != nil { return err } if ff.bp != nil { ff.bp.DiscardPage(pageID) } return nil } // WritePageHome 把页镜像写回它自己的位置 (WAL 提交时的"回填"步骤)。 // // 与 WritePageDirect 的唯一区别是【不动缓存副本】: 这份镜像是从缓存快照里来的, // 缓存里那一份的内容只会等于它或新于它, 无差别失效反而会丢掉提交窗口里写入线程刚 // 改出来的新内容 (并发写 + 提交时就是如此)。页的脏/干净状态交给提交方用 // File.MarkClean 按版本号统一收尾 —— 没被改过的转干净, 改过的继续留脏。 func (ff *File) WritePageHome(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) // ----------------------------------------------------------------------------------- // writeHeaderLocked 把头页 (4K) 落盘。调用方必须已持有 headerMu。 // 与 WriteHeader 的区别就是不重复加锁 —— 供 withHeader 内部使用, 避免自死锁。 func (ff *File) writeHeaderLocked(h *HeaderPage) error { var buf [PageSize4K]byte copy(buf[:], (*[PageSize4K]byte)(unsafe.Pointer(h))[:]) _, err := ff.f.WriteAt(buf[:], 0) if err == nil { ff.headerDirty = false // 头页已是最新 } return err } // withHeader 在 headerMu 保护下执行 fn。 // // fn 内可以安全地读 / 改 HeaderPage 的任何字段; 若还需要把头页落盘, 必须调用 // writeHeaderLocked (不能调用 WriteHeader —— 那会二次获取 headerMu, 直接死锁)。 // // fn 内【不要】调用会同步触发提交的动作 (典型是经 BufferPool 缓存的 WritePage: 脏页 // 超过阈值时它会在当前 goroutine 里直接调 Commit, 而 Commit 的 WAL 写入又要 headerMu): // 那会自死锁。需要这种写页请放到锁外, 见 allocPage。 func (ff *File) withHeader(fn func()) { ff.headerMu.Lock() defer ff.headerMu.Unlock() fn() } // WriteHeader 线程安全地把头页 (4K) 落盘。 func (ff *File) WriteHeader(h *HeaderPage) error { ff.headerMu.Lock() defer ff.headerMu.Unlock() return ff.writeHeaderLocked(h) } // FlushHeader 头页有未落盘的分配游标变更时把它写下去。 // // 兜底用途: 把缓冲池里的脏页直接刷盘之前必须先调用 (db.Close / 任何绕过 // Commit 的 Flush)—— 否则盘上会出现页号超过头页 NextFreePage 的页, 重开 // 后被二次分配覆盖。Commit 路径自己会写头页, 通常到这里已是干净的。 func (ff *File) FlushHeader(h *HeaderPage) error { ff.headerMu.Lock() defer ff.headerMu.Unlock() if !ff.headerDirty { return nil } return ff.writeHeaderLocked(h) } // allocPage 线程安全地分配一个页号: 在 headerMu 内摘空闲链 / 递增游标并推进头页, // 随后把头页落盘。 // // erase 决定是否把该页擦成空白页 (写缓存, 延迟落盘): // - 普通写入路径传 true: 复用的空闲页可能残留旧页类型, 必须擦干净, 否则按页号 // 读回来会被当成合法但类型错误的页。 // - WAL 打包路径传 false: WAL 页紧接着就在内存里被整页重造 (16KB 全写), 不需要 // 擦页; 更要紧的是此时正处在 Commit 内部 (持有 commitMu), 而擦页会走 // File.WritePage → BufferPool.WritePage, 脏页超阈值时同步回调 autoCommit → // 再进一次 Commit → 卡死在已持有的 commitMu 上 (fatal: all goroutines are asleep)。 // // 为什么 headerMu 只罩住"选号 + 推进头页", 擦页放在锁外: 擦页可能重入提交, // 而提交的 WAL 打包又要 headerMu —— 锁内擦页同样是自死锁。 // // 头页不在这里落盘 (只标脏): 分配的页本身只在缓冲池里, 未提交前盘上没有它们; // 头页游标的持久化收敛到 Commit / 空闲链变更 / Close 三处 (见 headerDirty 注释)。 func (ff *File) allocPage(h *HeaderPage, erase bool) uint32 { ff.headerMu.Lock() // writePage 传 nil: 只推进头页游标, 不做任何可能重入提交的写。 pid := h.newPageWithReader(ff.readPageDisk, nil) ff.headerDirty = true ff.headerMu.Unlock() if !erase { return pid } // 锁外擦页: 页头只留自述页号 (Type=0、UseSize=页头长), 其余字节全 0。 clean := make([]byte, PageSize) page.HdrOf(clean).Init(0, pid) _ = ff.WritePage(pid, clean) return pid } 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) }