TsDB2/DB/core/File.go

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)
}