File.go 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. package core
  2. import (
  3. "TsDB2/DB/core/pool"
  4. "os"
  5. "unsafe"
  6. )
  7. // File 只负责:打开、关闭、读、写 —— unsafe 全部收敛
  8. type File struct {
  9. f *os.File
  10. bp *pool.BufferPool // 页面缓冲池 (nil = 未启用)
  11. }
  12. func OpenFile(path string) (*File, error) {
  13. f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0644)
  14. if err != nil {
  15. return nil, err
  16. }
  17. return &File{f: f}, nil
  18. }
  19. func (ff *File) Close() error {
  20. if ff.bp != nil {
  21. ff.bp.Close() // 先刷盘再关文件
  22. }
  23. return ff.f.Close()
  24. }
  25. // InitBufferPool 初始化页面缓冲池, 拦截 ReadPage/WritePage 实现内存缓存
  26. // capacity: 最大缓存页数 (建议 10000~50000)
  27. func (ff *File) InitBufferPool(capacity int) {
  28. ff.bp = pool.NewBufferPool(capacity, ff.readPageDisk, ff.writePageDisk)
  29. }
  30. // SetDirtyThreshold 设置脏页阈值, 超过时触发回调 (配合 WAL 自动 Commit)
  31. func (ff *File) SetDirtyThreshold(threshold int, onThreshold func()) {
  32. if ff.bp != nil {
  33. ff.bp.SetDirtyThreshold(threshold, onThreshold)
  34. }
  35. }
  36. // SetOnPageWriteCallback 设置页写入回调(用于事务缓存原始数据)
  37. func (ff *File) SetOnPageWriteCallback(callback func(pageID uint32)) {
  38. if ff.bp != nil {
  39. ff.bp.SetOnPageWriteCallback(callback)
  40. }
  41. }
  42. // LockWrites 锁定写入(禁止 Insert/Update/Delete)
  43. func (ff *File) LockWrites() {
  44. if ff.bp != nil {
  45. ff.bp.LockWrites()
  46. }
  47. }
  48. // UnlockWrites 解锁写入(允许 Insert/Update/Delete)
  49. func (ff *File) UnlockWrites() {
  50. if ff.bp != nil {
  51. ff.bp.UnlockWrites()
  52. }
  53. }
  54. // SetBufferPoolCapacity 设置 BufferPool 容量(页数)
  55. func (ff *File) SetBufferPoolCapacity(capacity int) {
  56. if ff.bp != nil {
  57. ff.bp.SetCapacity(capacity)
  58. }
  59. }
  60. // CollectDirty 收集所有脏页快照并清除脏标记 (配合 WAL 使用)
  61. func (ff *File) CollectDirty() []pool.DirtyPage {
  62. if ff.bp != nil {
  63. return ff.bp.CollectDirty()
  64. }
  65. return nil
  66. }
  67. // FlushBufferPool 刷盘 (仅在 Close 时使用)
  68. func (ff *File) FlushBufferPool() {
  69. if ff.bp != nil {
  70. ff.bp.Flush()
  71. }
  72. }
  73. func (ff *File) WritePage16K(pageID uint32, buf *[PageSize]byte) error {
  74. // 头占一个4K,所以 pageID 从 1 开始
  75. offset := int64(PageSize4K) + int64(pageID-1)*int64(PageSize)
  76. _, err := ff.f.WriteAt(buf[:], offset)
  77. return err
  78. }
  79. // -----------------------------------------------------------------------------------
  80. // ✅ 动态长度:索引页 — 缓冲池拦截层
  81. // -----------------------------------------------------------------------------------
  82. func (ff *File) WritePage(pageID uint32, buf []byte) error {
  83. if ff.bp != nil {
  84. return ff.bp.WritePage(pageID, buf) // 写缓存, 延迟落盘
  85. }
  86. return ff.writePageDisk(pageID, buf)
  87. }
  88. func (ff *File) ReadPage(pageID uint32, buf []byte) error {
  89. if ff.bp != nil {
  90. return ff.bp.ReadPage(pageID, buf) // 先查缓存
  91. }
  92. return ff.readPageDisk(pageID, buf)
  93. }
  94. // WritePageDirect 直接写盘 (绕过 BufferPool, 用于 WAL 回写)
  95. func (ff *File) WritePageDirect(pageID uint32, buf []byte) error {
  96. return ff.writePageDisk(pageID, buf)
  97. }
  98. // SyncCache 同步缓存页 (WAL 提交后调用, 保持缓存与磁盘一致)
  99. func (ff *File) SyncCache(pageID uint32, data []byte) {
  100. if ff.bp != nil {
  101. ff.bp.UpdatePage(pageID, data)
  102. }
  103. }
  104. // ClearCache 清空所有缓存 (WAL 提交后调用)
  105. func (ff *File) ClearCache() {
  106. if ff.bp != nil {
  107. ff.bp.ClearAll()
  108. }
  109. }
  110. // -----------------------------------------------------------------------------------
  111. // ✅ 底层磁盘 I/O (缓冲池回源)
  112. // -----------------------------------------------------------------------------------
  113. func (ff *File) writePageDisk(pageID uint32, buf []byte) error {
  114. offset := int64(PageSize4K) + int64(pageID-1)*int64(len(buf))
  115. _, err := ff.f.WriteAt(buf, offset)
  116. return err
  117. }
  118. func (ff *File) readPageDisk(pageID uint32, buf []byte) error {
  119. offset := int64(PageSize4K) + int64(pageID-1)*int64(len(buf))
  120. _, err := ff.f.ReadAt(buf, offset)
  121. return err
  122. }
  123. // -----------------------------------------------------------------------------------
  124. // ✅ 头读写(固定4K)
  125. // -----------------------------------------------------------------------------------
  126. func (ff *File) WriteHeader(h *HeaderPage) error {
  127. var buf [PageSize4K]byte
  128. copy(buf[:], (*[PageSize4K]byte)(unsafe.Pointer(h))[:])
  129. _, err := ff.f.WriteAt(buf[:], 0)
  130. return err
  131. }
  132. func (ff *File) ReadHeader() (*HeaderPage, error) {
  133. buf := make([]byte, PageSize4K)
  134. _, err := ff.f.ReadAt(buf, 0)
  135. if err != nil {
  136. return nil, err
  137. }
  138. h := &HeaderPage{}
  139. copy((*[PageSize4K]byte)(unsafe.Pointer(h))[:], buf)
  140. return h, nil
  141. }
  142. func (ff *File) TruncateSize(size int64) error {
  143. return ff.f.Truncate(size)
  144. }