package core import ( "TsDB2/DB/core/page" ) // ============================================================================= // SchemaExtAllocator —— 全库共享的 schema 扩展页分配器 // ============================================================================= // // 为什么共享: schema 一般只有几十~几千字节。若每张表独占一张 16KB 页, 表一多 // 就极其浪费; 全库共用一条链, 大家按实际长度挤在同一张页里。 // // 页链与定位: // // 表页第 1 页 : [PageHdr][SchemaExtPageID u32][TableInfo][内联块]... // └─ 链头 (0 = 还没有扩展页) // 扩展页 : [PageHdr][frag][frag]...[空闲] ─NextPage─> ... // frag : [SchemaFragHdr 8B][payload] // // 每张表自己的首个续块位置记在它的 TableInfo.ExtPage/ExtOff 里 (不在这里)。 // 页头的 UseSize 复用为「尾页 bump 偏移」, 重启后据此恢复 lastOff。 // // 回收策略: 只往尾页追加(bump), 续块不单独释放 —— 每次 writeSchema 都是整条重写, // 旧续块立刻成为垃圾。当尾页放不下时: // - 若压缩重排后能减少页数 → Compact(): 把所有活续块紧排进新页, 原子换头, 再回收旧链 // - 否则直接追加新页 // // 这样做无需页内 free list 与碎片管理, 代价是压缩时要把所有表重写一遍 (只在 DDL 触发)。 type SchemaExtAllocator struct { handle page.HandleDb tables func() []*Table // 取全量表, 供压缩重排使用 saveHead func(pageID uint32) error // 把链头写回表页第 1 页 firstPID uint32 // 链首页 (0 = 空) lastPID uint32 // 链尾页, 追加位置 lastOff uint32 // 尾页已 bump 到的物理偏移 pages int // 链长度 inited bool compacting bool // 正在压缩重排 → 期间禁止再压缩 } // schemaExtHeadSize 表页第 1 页在 PageHdr 之后为 SchemaExtPageID 预留的字节数 // (所有表页统一预留, 使表偏移遍历逻辑无需区分第几页) const schemaExtHeadSize = 4 // schemaFragMaxPayload 单个续块能承载的最大 payload func schemaFragMaxPayload() int { return int(PageSize) - int(page.PageHdrSize) - schemaFragHdrSize } // schemaFrag 一次分配得到的续块位置与容量 type schemaFrag struct { PageID uint32 Off uint32 Len uint16 // payload 字节数 } // newSchemaExtAllocator 创建分配器 (firstPID 为已存在的链头, 0 表示空) func newSchemaExtAllocator(handle page.HandleDb, firstPID uint32, tables func() []*Table, saveHead func(uint32) error) *SchemaExtAllocator { return &SchemaExtAllocator{ handle: handle, tables: tables, saveHead: saveHead, firstPID: firstPID, } } // init 定位链尾并恢复 bump 偏移 (懒加载, 首次分配时才读盘) func (a *SchemaExtAllocator) init() error { if a.inited { return nil } a.inited = true a.pages = 0 a.lastPID = 0 a.lastOff = 0 for pid := a.firstPID; pid != 0; { buf := make([]byte, PageSize) if err := a.handle.ReadPage(pid, buf); err != nil { return err } hdr := page.HdrOf(buf) a.pages++ if hdr.NextPage == 0 { a.lastPID = pid if hdr.UseSize >= page.PageHdrSize && hdr.UseSize <= PageSize { a.lastOff = hdr.UseSize } else { a.lastOff = page.PageHdrSize } return nil } pid = hdr.NextPage } return nil } // flushTail 把尾页的 bump 偏移写回页头 UseSize func (a *SchemaExtAllocator) flushTail() error { if a.lastPID == 0 { return nil } buf := make([]byte, PageSize) if err := a.handle.ReadPage(a.lastPID, buf); err != nil { return err } page.HdrOf(buf).UseSize = a.lastOff return a.handle.SavePage(a.lastPID, buf) } // appendPage 在链尾追加一张新的扩展页, 并把它挂到前驱的 NextPage 上 func (a *SchemaExtAllocator) appendPage() error { prev := a.lastPID if prev != 0 { // 换页前先把旧尾页的 bump 偏移落盘 if err := a.flushTail(); err != nil { return err } } pid := a.handle.NewPage() buf := make([]byte, PageSize) hdr := page.HdrOf(buf) hdr.Type = page.SchemaExtIndex hdr.PageID = pid hdr.UseSize = page.PageHdrSize hdr.PrePage = prev if err := a.handle.SavePage(pid, buf); err != nil { return err } if prev != 0 { pb := make([]byte, PageSize) if err := a.handle.ReadPage(prev, pb); err != nil { return err } page.HdrOf(pb).NextPage = pid if err := a.handle.SavePage(prev, pb); err != nil { return err } } else { // 空链 → 新页即链头, 需要回写表页第 1 页 a.firstPID = pid if a.saveHead != nil { if err := a.saveHead(pid); err != nil { return err } } } a.lastPID = pid a.lastOff = page.PageHdrSize a.pages++ return nil } // AllocChain 为 total 字节的续块数据分配若干片段, 返回每段的位置与容量 // 分配只发生在尾页, 一段放不下就跨页续下一段 // // 注意: 压缩重排会回收整条旧链, 使**已经记下的片段地址全部失效**。所以压缩只允许 // 发生在开始记录地址之前 (下面这一次判断); 之后的循环一律只追加新页, 绝不压缩。 func (a *SchemaExtAllocator) AllocChain(total int) ([]schemaFrag, error) { if total <= 0 { return nil, nil } if err := a.init(); err != nil { return nil, err } // 尾页装不下这次要写的全部数据 → 马上要新增页, 这是值得压缩重排的时机。 // 反过来, 只要尾页装得下就不压缩 —— 否则每次 DDL 都要重排所有表。 if a.lastPID != 0 && PageSize-a.lastOff < uint32(total) { if a.shouldCompact(total) { if err := a.Compact(); err != nil { return nil, err } } } frags := make([]schemaFrag, 0, 1+total/schemaFragMaxPayload()) remain := total for remain > 0 { if a.lastPID == 0 || PageSize-a.lastOff < uint32(schemaFragHdrSize)+1 { if err := a.appendPage(); err != nil { return nil, err } } n := int(PageSize-a.lastOff) - schemaFragHdrSize if n > remain { n = remain } if n > schemaFragMaxPayload() { n = schemaFragMaxPayload() } frags = append(frags, schemaFrag{PageID: a.lastPID, Off: a.lastOff, Len: uint16(n)}) a.lastOff += uint32(schemaFragHdrSize + n) remain -= n } if err := a.flushTail(); err != nil { return nil, err } return frags, nil } // liveStats 统计当前所有表续块占用的净字节数与续块数 func (a *SchemaExtAllocator) liveStats() (int, int) { live, n := 0, 0 if a.tables == nil { return 0, 0 } for _, t := range a.tables() { if t == nil || t.info.ExtPage == 0 { continue } inline := int(t.info.FieldAllocateSize) data := t.serializeSchema() if len(data) <= inline { continue } live += len(data) - inline n++ } return live, n } // shouldCompact 判断压缩重排 (外加即将写入的 extra 字节) 是否能减少页数 (不能减就不折腾) func (a *SchemaExtAllocator) shouldCompact(extra int) bool { if a.pages <= 1 || a.compacting { return false } live, liveTables := a.liveStats() usable := uint64(PageSize - page.PageHdrSize) need := uint64(live) + uint64(liveTables)*uint64(schemaFragHdrSize) + uint64(extra) pagesAfter := int((need + usable - 1) / usable) if pagesAfter < 1 { pagesAfter = 1 } return pagesAfter < a.pages } // Compact 满页压缩重排: 把所有活续块紧排进一条新链, 换头后再回收旧链 // // 先在新链上把所有表重写完毕, 最后才回写表页里的链头 —— 中途失败只会泄漏几张 // 页, 不会丢失已有 schema。 func (a *SchemaExtAllocator) Compact() error { if err := a.init(); err != nil { return err } oldFirst := a.firstPID // 新链在独立的临时分配器上构建, saveHead 置空 → 先不暴露给外部 // compacting=true → 构建过程中禁止再压缩 (否则会递归, 且新链地址会失效) tmp := &SchemaExtAllocator{ handle: a.handle, tables: a.tables, saveHead: func(uint32) error { return nil }, inited: true, compacting: true, } if err := tmp.appendPage(); err != nil { return err } for _, t := range a.tables() { if t == nil || t.info.ExtPage == 0 { continue } if err := t.writeSchemaExt(tmp); err != nil { return err } } // 原子换头 a.firstPID, a.lastPID, a.lastOff, a.pages = tmp.firstPID, tmp.lastPID, tmp.lastOff, tmp.pages if a.saveHead != nil { if err := a.saveHead(a.firstPID); err != nil { return err } } a.freeChain(oldFirst) return nil } // freeChain 回收整条扩展页链 func (a *SchemaExtAllocator) freeChain(first uint32) { if a.handle.FreePage == nil { return } for pid := first; pid != 0; { buf := make([]byte, PageSize) if err := a.handle.ReadPage(pid, buf); err != nil { return } next := page.HdrOf(buf).NextPage a.handle.FreePage(pid) pid = next } }