TsDB2/DB/core/File.go

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