schema_ext.go 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301
  1. package core
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core/page"
  4. )
  5. // =============================================================================
  6. // SchemaExtAllocator —— 全库共享的 schema 扩展页分配器
  7. // =============================================================================
  8. //
  9. // 为什么共享: schema 一般只有几十~几千字节。若每张表独占一张 16KB 页, 表一多
  10. // 就极其浪费; 全库共用一条链, 大家按实际长度挤在同一张页里。
  11. //
  12. // 页链与定位:
  13. //
  14. // 表页第 1 页 : [PageHdr][SchemaExtPageID u32][TableInfo][内联块]...
  15. // └─ 链头 (0 = 还没有扩展页)
  16. // 扩展页 : [PageHdr][frag][frag]...[空闲] ─NextPage─> ...
  17. // frag : [SchemaFragHdr 8B][payload]
  18. //
  19. // 每张表自己的首个续块位置记在它的 TableInfo.ExtPage/ExtOff 里 (不在这里)。
  20. // 页头的 UseSize 复用为「尾页 bump 偏移」, 重启后据此恢复 lastOff。
  21. //
  22. // 回收策略: 只往尾页追加(bump), 续块不单独释放 —— 每次 writeSchema 都是整条重写,
  23. // 旧续块立刻成为垃圾。当尾页放不下时:
  24. // - 若压缩重排后能减少页数 → Compact(): 把所有活续块紧排进新页, 原子换头, 再回收旧链
  25. // - 否则直接追加新页
  26. //
  27. // 这样做无需页内 free list 与碎片管理, 代价是压缩时要把所有表重写一遍 (只在 DDL 触发)。
  28. type SchemaExtAllocator struct {
  29. handle page.HandleDb
  30. tables func() []*Table // 取全量表, 供压缩重排使用
  31. saveHead func(pageID uint32) error // 把链头写回表页第 1 页
  32. firstPID uint32 // 链首页 (0 = 空)
  33. lastPID uint32 // 链尾页, 追加位置
  34. lastOff uint32 // 尾页已 bump 到的物理偏移
  35. pages int // 链长度
  36. inited bool
  37. compacting bool // 正在压缩重排 → 期间禁止再压缩
  38. }
  39. // schemaExtHeadSize 表页第 1 页在 PageHdr 之后为 SchemaExtPageID 预留的字节数
  40. // (所有表页统一预留, 使表偏移遍历逻辑无需区分第几页)
  41. const schemaExtHeadSize = 4
  42. // schemaFragMaxPayload 单个续块能承载的最大 payload
  43. func schemaFragMaxPayload() int {
  44. return int(PageSize) - int(page.PageHdrSize) - schemaFragHdrSize
  45. }
  46. // schemaFrag 一次分配得到的续块位置与容量
  47. type schemaFrag struct {
  48. PageID uint32
  49. Off uint32
  50. Len uint16 // payload 字节数
  51. }
  52. // newSchemaExtAllocator 创建分配器 (firstPID 为已存在的链头, 0 表示空)
  53. func newSchemaExtAllocator(handle page.HandleDb, firstPID uint32,
  54. tables func() []*Table, saveHead func(uint32) error) *SchemaExtAllocator {
  55. return &SchemaExtAllocator{
  56. handle: handle,
  57. tables: tables,
  58. saveHead: saveHead,
  59. firstPID: firstPID,
  60. }
  61. }
  62. // init 定位链尾并恢复 bump 偏移 (懒加载, 首次分配时才读盘)
  63. func (a *SchemaExtAllocator) init() error {
  64. if a.inited {
  65. return nil
  66. }
  67. a.inited = true
  68. a.pages = 0
  69. a.lastPID = 0
  70. a.lastOff = 0
  71. for pid := a.firstPID; pid != 0; {
  72. buf := make([]byte, PageSize)
  73. if err := a.handle.ReadPage(pid, buf); err != nil {
  74. return err
  75. }
  76. hdr := page.HdrOf(buf)
  77. a.pages++
  78. if hdr.NextPage == 0 {
  79. a.lastPID = pid
  80. if hdr.UseSize >= page.PageHdrSize && hdr.UseSize <= PageSize {
  81. a.lastOff = hdr.UseSize
  82. } else {
  83. a.lastOff = page.PageHdrSize
  84. }
  85. return nil
  86. }
  87. pid = hdr.NextPage
  88. }
  89. return nil
  90. }
  91. // flushTail 把尾页的 bump 偏移写回页头 UseSize
  92. func (a *SchemaExtAllocator) flushTail() error {
  93. if a.lastPID == 0 {
  94. return nil
  95. }
  96. buf := make([]byte, PageSize)
  97. if err := a.handle.ReadPage(a.lastPID, buf); err != nil {
  98. return err
  99. }
  100. page.HdrOf(buf).UseSize = a.lastOff
  101. return a.handle.SavePage(a.lastPID, buf)
  102. }
  103. // appendPage 在链尾追加一张新的扩展页, 并把它挂到前驱的 NextPage 上
  104. func (a *SchemaExtAllocator) appendPage() error {
  105. prev := a.lastPID
  106. if prev != 0 {
  107. // 换页前先把旧尾页的 bump 偏移落盘
  108. if err := a.flushTail(); err != nil {
  109. return err
  110. }
  111. }
  112. pid := a.handle.NewPage()
  113. buf := make([]byte, PageSize)
  114. hdr := page.HdrOf(buf)
  115. hdr.Type = page.SchemaExtIndex
  116. hdr.PageID = pid
  117. hdr.UseSize = page.PageHdrSize
  118. hdr.PrePage = prev
  119. if err := a.handle.SavePage(pid, buf); err != nil {
  120. return err
  121. }
  122. if prev != 0 {
  123. pb := make([]byte, PageSize)
  124. if err := a.handle.ReadPage(prev, pb); err != nil {
  125. return err
  126. }
  127. page.HdrOf(pb).NextPage = pid
  128. if err := a.handle.SavePage(prev, pb); err != nil {
  129. return err
  130. }
  131. } else {
  132. // 空链 → 新页即链头, 需要回写表页第 1 页
  133. a.firstPID = pid
  134. if a.saveHead != nil {
  135. if err := a.saveHead(pid); err != nil {
  136. return err
  137. }
  138. }
  139. }
  140. a.lastPID = pid
  141. a.lastOff = page.PageHdrSize
  142. a.pages++
  143. return nil
  144. }
  145. // AllocChain 为 total 字节的续块数据分配若干片段, 返回每段的位置与容量
  146. // 分配只发生在尾页, 一段放不下就跨页续下一段
  147. //
  148. // 注意: 压缩重排会回收整条旧链, 使**已经记下的片段地址全部失效**。所以压缩只允许
  149. // 发生在开始记录地址之前 (下面这一次判断); 之后的循环一律只追加新页, 绝不压缩。
  150. func (a *SchemaExtAllocator) AllocChain(total int) ([]schemaFrag, error) {
  151. if total <= 0 {
  152. return nil, nil
  153. }
  154. if err := a.init(); err != nil {
  155. return nil, err
  156. }
  157. // 尾页装不下这次要写的全部数据 → 马上要新增页, 这是值得压缩重排的时机。
  158. // 反过来, 只要尾页装得下就不压缩 —— 否则每次 DDL 都要重排所有表。
  159. if a.lastPID != 0 && PageSize-a.lastOff < uint32(total) {
  160. if a.shouldCompact(total) {
  161. if err := a.Compact(); err != nil {
  162. return nil, err
  163. }
  164. }
  165. }
  166. frags := make([]schemaFrag, 0, 1+total/schemaFragMaxPayload())
  167. remain := total
  168. for remain > 0 {
  169. if a.lastPID == 0 || PageSize-a.lastOff < uint32(schemaFragHdrSize)+1 {
  170. if err := a.appendPage(); err != nil {
  171. return nil, err
  172. }
  173. }
  174. n := int(PageSize-a.lastOff) - schemaFragHdrSize
  175. if n > remain {
  176. n = remain
  177. }
  178. if n > schemaFragMaxPayload() {
  179. n = schemaFragMaxPayload()
  180. }
  181. frags = append(frags, schemaFrag{PageID: a.lastPID, Off: a.lastOff, Len: uint16(n)})
  182. a.lastOff += uint32(schemaFragHdrSize + n)
  183. remain -= n
  184. }
  185. if err := a.flushTail(); err != nil {
  186. return nil, err
  187. }
  188. return frags, nil
  189. }
  190. // liveStats 统计当前所有表续块占用的净字节数与续块数
  191. func (a *SchemaExtAllocator) liveStats() (int, int) {
  192. live, n := 0, 0
  193. if a.tables == nil {
  194. return 0, 0
  195. }
  196. for _, t := range a.tables() {
  197. if t == nil || t.info.ExtPage == 0 {
  198. continue
  199. }
  200. inline := int(t.info.FieldAllocateSize)
  201. data := t.serializeSchema()
  202. if len(data) <= inline {
  203. continue
  204. }
  205. live += len(data) - inline
  206. n++
  207. }
  208. return live, n
  209. }
  210. // shouldCompact 判断压缩重排 (外加即将写入的 extra 字节) 是否能减少页数 (不能减就不折腾)
  211. func (a *SchemaExtAllocator) shouldCompact(extra int) bool {
  212. if a.pages <= 1 || a.compacting {
  213. return false
  214. }
  215. live, liveTables := a.liveStats()
  216. usable := uint64(PageSize - page.PageHdrSize)
  217. need := uint64(live) + uint64(liveTables)*uint64(schemaFragHdrSize) + uint64(extra)
  218. pagesAfter := int((need + usable - 1) / usable)
  219. if pagesAfter < 1 {
  220. pagesAfter = 1
  221. }
  222. return pagesAfter < a.pages
  223. }
  224. // Compact 满页压缩重排: 把所有活续块紧排进一条新链, 换头后再回收旧链
  225. //
  226. // 先在新链上把所有表重写完毕, 最后才回写表页里的链头 —— 中途失败只会泄漏几张
  227. // 页, 不会丢失已有 schema。
  228. func (a *SchemaExtAllocator) Compact() error {
  229. if err := a.init(); err != nil {
  230. return err
  231. }
  232. oldFirst := a.firstPID
  233. // 新链在独立的临时分配器上构建, saveHead 置空 → 先不暴露给外部
  234. // compacting=true → 构建过程中禁止再压缩 (否则会递归, 且新链地址会失效)
  235. tmp := &SchemaExtAllocator{
  236. handle: a.handle,
  237. tables: a.tables,
  238. saveHead: func(uint32) error { return nil },
  239. inited: true,
  240. compacting: true,
  241. }
  242. if err := tmp.appendPage(); err != nil {
  243. return err
  244. }
  245. for _, t := range a.tables() {
  246. if t == nil || t.info.ExtPage == 0 {
  247. continue
  248. }
  249. if err := t.writeSchemaExt(tmp); err != nil {
  250. return err
  251. }
  252. }
  253. // 原子换头
  254. a.firstPID, a.lastPID, a.lastOff, a.pages = tmp.firstPID, tmp.lastPID, tmp.lastOff, tmp.pages
  255. if a.saveHead != nil {
  256. if err := a.saveHead(a.firstPID); err != nil {
  257. return err
  258. }
  259. }
  260. a.freeChain(oldFirst)
  261. return nil
  262. }
  263. // freeChain 回收整条扩展页链
  264. func (a *SchemaExtAllocator) freeChain(first uint32) {
  265. if a.handle.FreePage == nil {
  266. return
  267. }
  268. for pid := first; pid != 0; {
  269. buf := make([]byte, PageSize)
  270. if err := a.handle.ReadPage(pid, buf); err != nil {
  271. return
  272. }
  273. next := page.HdrOf(buf).NextPage
  274. a.handle.FreePage(pid)
  275. pid = next
  276. }
  277. }