boltdb
南风未起 2025-07-22 19:39:22 etcd
# 1.page
# 1.1.简介
BoltDB是一个用Go语言编写的嵌入式键值存储数据库，支持完整的ACID事务，所有的数据存储在一个文件。BoltDB维护键的前缀映射，范围查询时可以快速定位磁盘上的B+树组织数据。BoltDB特别适用于轻量级、高性能、易于部署和维护的数据库解决方案的场景。--- 特点 1.每个db对应一个文件，文件按照page size(4096 bytes)划分page 2.前两个page保存metadata，特殊的page保存freelist(存放空闲page的id),剩下的page按照B+树组织 3.B+树的每个节点对应一个或多个连续page 4.基于mmap创建共享、只读的文件映射,调用madvise(MADV_RANDOM)由操作系统管理page cache 5.不涉及WAL(落盘前的暂存日志),只有commit时才写到磁盘 6.commit时将dirty page写入新的page,从而保证同时读的事务不受影响1
2
3
4
5
6
7
# 1.2.物理页
boltdb涉及到数据落盘，基于页page操作文件，采用mmap在内存映射操作文件，一个文件拆分为多页，以提高执行效率。每页由两部分数据构成：页头数据+真实数据，页头信息占16字节，页大小与操作系统对应的内存页大小一致——4k。
type pgid uint64 // 页ID const pageHeaderSize = int(unsafe.Offsetof(((*page)(nil)).ptr)) // 页头大小 const minKeysPerPage = 2 // 每页最少key const ( branchPageFlag = 0x01 // 分支页,用于存储索引页和每页起始key leafPageFlag = 0x02 // 叶子页,存储真实key/value metaPageFlag = 0x04 // 元数据页,保存元数据,包括空闲列表pgId、桶的根页 freelistPageFlag = 0x10 // 空闲页,记录空闲的及待释放的页信息,用于后续分配空间时的页复用 ) type page struct { id pgid // pageId,8字节 flags uint16 // page类型,2字节 count uint16 // page元素数量,2字节 overflow uint32 // 数据是否溢出,用于空闲列表,4字节 }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19page前16个字节保存页头信息，数据区ptr是一个无类型指针，记录每页中真实存储的数据地址。
# 1.3.元数据
物理页
page实现了meta()方法，类型属于元数据页时可以通过该方法获取具体的元数据信息，起始就是page跳过页头的ptr指向数据信息。func (p *page) meta() *meta { return (*meta)(unsafeAdd(unsafe.Pointer(p), unsafe.Sizeof(*p))) }1
2
3meta记录空闲页、root bucket等信息，数据操作时根据root bucket找到整个B+树的入口，然后才能进行查找、更新操作。type meta struct { magic uint32 // boltdb的魔数 version uint32 // boltdb的版本号 pageSize uint32 // page大小,与操作系统默认的页大小保持一致 flags uint32 // 页类型,保留值 root bucket // 根节点,作为总入口 freelist pgid // 空闲页ID pgid pgid // 下一个待分配的pgID txid txid // 下一个待分配的txID checksum uint64 // 用作校验的校验和 }1
2
3
4
5
6
7
8
9
10
11
// meta数据拷贝到page func (m *meta) copy(dest *meta) { *dest = *m } // 计算校验和 func (m *meta) sum64() uint64 { var h = fnv.New64a() // 写入校验参数 _, _ = h.Write((*[unsafe.Offsetof(meta{}.checksum)]byte)(unsafe.Pointer(m))[:]) return h.Sum64() } // 把meta写入page func (m *meta) write(p *page) { ... // 指定pgID和类型 p.id = pgid(m.txid % 2) p.flags |= metaPageFlag // 计算校验和 m.checksum = m.sum64() // 将meta数据复制到page(p.meta()拿到的是偏移页头后的数据地址) m.copy(p.meta()) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26meta在db文件中有两个，存储着同一份数据，避免一个meta写坏时无参考对象进行恢复。此外，meta.root会作为B+树的总入口存在，其保存的是树的根节点。
更具体来说，
meta page采用副本机制，两个meta page轮流刷盘持久化，记录最新的数据状态，每轮事务初始化拷贝txid较大的meta page作为新的meta副本。
# 1.4.空闲页
freelist在db文件中只有一个，其pageID不会固定在pgID 2的位置，随着数据的写入，freelist也会进行变更，pgID会发生变化。freelist维护空闲的pgID集，需要分配pgID时会先从空闲页找，无法找到会对db文件进行扩充，得到一个新的pgID。type pidSet map[pgid]struct{} // 相同连续页数的页开始ID集合 type FreelistType string // freelist后端类型,用字符串表示不同的实现策略 const ( FreelistArrayType = FreelistType("array") // 数组类型,适用于按序访问空闲页场景 FreelistMapType = FreelistType("hashmap") // 哈希映射类型,提供更快的查找速度,适用于频繁、随机访问情况 ) type freelist struct { freelistType FreelistType // freelist类型,涉及不同策略实现 ids []pgid // 存储所有已释放且可供分配的页ID allocs map[pgid]txid // 记录事务ID分配到的特定页面ID pending map[txid]*txPending // 记录即将释放的页ID和所属的事务ID cache map[pgid]struct{} // 快速查找所有空闲和待处理页面ID的缓存 freemaps map[uint64]pidSet // 连续页大小分类的空闲页,键:连续页大小,值:相同大小起始页ID集 forwardMap map[pgid]uint64 // 正向映射,起始页ID-->span大小 backwardMap map[pgid]uint64 // 反向映射,结束页ID-->span大小 allocate func(txid txid, n int) pgid // 分配函数,根据提供的事务ID和需求的页面数量分配页面 free_count func() int // 空闲页面计数函数 mergeSpans func(ids pgids) // 合并连续空闲页面的函数 getFreePageIDs func() []pgid // 获取所有空闲页面ID的函数 readIDs func(pgids []pgid) // 读取一系列页面ID并初始化freelist函数 }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23boltdb采用采用mmap操作时，只会扩容不会缩容，也就是只借不还，因此读写事务释放的page会缓存到空闲列表统一管理，不会将其归还给操作系统，避免page位于page buffer时打破连续性。
空闲列表可以转换为页信息写到磁盘，由于
pageHeader中的count字段是一个uint16，只有两个字节，因此空闲页的数量大小影响count的存储位置。数量溢出时，page.ptr数据区的第一个字节会用来存储空闲页数量，同时将page.count设置为0xFFFF标记。// 将所有空闲的ID和待处理的ID复制到一个排序后列表 func (f *freelist) copyall(dst []pgid) { // 待释放pgID m := make(pgids, 0, f.pending_count()) for _, txp := range f.pending { m = append(m, txp.ids...) } // pgID排序 sort.Sort(m) // 排序合并以释放pgID mergepgids(dst, f.getFreePageIDs(), m) } // 空闲页转换物理页 func (f *freelist) write(p *page) error { // 页头设置页标识 p.flags |= freelistPageFlag // 空闲页处理 l := f.count() if l == 0 { p.count = uint16(l) } else if l < 0xFFFF { // 数量小于0xFFFF,空闲页数量写入页头 p.count = uint16(l) var ids []pgid // 偏移到page.ptr地址 data := unsafeAdd(unsafe.Pointer(p), unsafe.Sizeof(*p)) // 开辟一个slice存储id,地址赋给page.ptr unsafeSlice(unsafe.Pointer(&ids), data, l) // 空闲页pgID写入地址 f.copyall(ids) } else { // 数量大于0xFFFF,标记page.count p.count = 0xFFFF var ids []pgid // 偏移到page.ptr地址 data := unsafeAdd(unsafe.Pointer(p), unsafe.Sizeof(*p)) // pgID切片地址赋给page.ptr unsafeSlice(unsafe.Pointer(&ids), data, l+1) // 空闲页数量记录到第一个字节 ids[0] = pgid(l) // 已释放pgID和待释放pgID有序合并到ids f.copyall(ids[1:]) } return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48物理页中的空闲页信息可以加载为
freelist结构，转换时会根据page.count决定加载方式。// 从物理页初始化freelist func (f *freelist) read(p *page) { ... // 获取page.count var idx, count = 0, int(p.count) // 数量溢出 if count == 0xFFFF { // 指向page.ptr第一个字节 idx = 1 // 获取page.ptr地址 c := *(*pgid)(unsafeAdd(unsafe.Pointer(p), unsafe.Sizeof(*p))) // 读取第一个字节字节数量 count = int(c) } // 没有pgid if count == 0 { f.ids = nil } else { var ids []pgid // 偏移到page.ptr[1:] data := unsafeIndex(unsafe.Pointer(p), unsafe.Sizeof(*p), unsafe.Sizeof(ids[0]), idx) // ids切片地址赋给page.ptr[1:] unsafeSlice(unsafe.Pointer(&ids), data, count) // 复制物理页中的pgIDs idsCopy := make([]pgid, count) copy(idsCopy, ids) // 排序 sort.Sort(pgids(idsCopy)) // 排序后的ID列表读入freelist结构 f.readIDs(idsCopy) } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33读写事务提交时，基于
copy-on-write机制，针对所有涉及更改的节点分配新的page副本，此时会尝试从freelist申请，无法申请时会进行新一轮mmap操作拓展db页容量。// hashmapAllocate实现类型,基于分配数量从map找是否有满足块 // 申请n个连续页,返回起始pgID func (f *freelist) arrayAllocate(txid txid, n int) pgid { // 无空闲页分配 if len(f.ids) == 0 { return 0 } var initial, previd pgid // 遍历空闲pgID for i, id := range f.ids { // 不连续,重新初始化起始pgID if previd == 0 || id-previd != 1 { initial = id } // 找到连续的块,返回 if (id-initial)+1 == pgid(n) { // 找到前n个连续块 if (i + 1) == n { f.ids = f.ids[i+1:] } else { // 右侧未未选中pgID左移n copy(f.ids[i-n+1:], f.ids[i+1:]) // 更新剩余pgID f.ids = f.ids[:len(f.ids)-n] } // 更新缓存 for i := pgid(0); i < pgid(n); i++ { delete(f.cache, initial+i) } // 记录已分配的page f.allocs[initial] = txid // 返回起始pgID return initial } previd = id } return 0 }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# 1.5.分支页
branch page用于存储数据索引，对应B+树中枝叶节点，由于boltdb存储的key不定长，因此branch page采取shadow paging技术，page body中会放一系列定长的branchPageElement，标注key的地址、长度及映射的pgID。// branchPageElement represents a node on a branch page. type branchPageElement struct { pos uint32 // 数据偏移量 ksize uint32 // key的长度 pgid pgid // 指向的pageId } // branchPageElements retrieves a list of branch nodes. func (p *page) branchPageElements() []branchPageElement { if p.count == 0 { return nil } var elems []branchPageElement // 偏移到page body data := unsafeAdd(unsafe.Pointer(p), unsafe.Sizeof(*p)) // page body中的branchElement放到切片 unsafeSlice(unsafe.Pointer(&elems), data, int(p.count)) return elems }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 1.6.叶子页
leaf page用于存储key/value数据，对应B+树的叶子节点，由于key存储的不定长，因此leaf page采取shadow paging技术，page body添加定长的leafPageElement，标注key地址、长度以及value长度。type leafPageElement struct { flags uint32 // 标识子bucket pos uint32 // 数据偏移量 ksize uint32 // key的长度 vsize uint32 // value的长度 } // leafPageElements retrieves a list of leaf nodes. func (p *page) leafPageElements() []leafPageElement { if p.count == 0 { return nil } var elems []leafPageElement // 偏移到page body data := unsafeAdd(unsafe.Pointer(p), unsafe.Sizeof(*p)) // page body中leafPageElement元素转入切片 unsafeSlice(unsafe.Pointer(&elems), data, int(p.count)) return elems }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 1.6.树节点
B+树中的基本构成单元是node，page对应文件系统的物理页，node则是page在内存的抽象，包括branch node和leaf node两种类型。leaf node没有children信息及key信息；branch node会记录key信息，children不一定为空。type inodes []inode type inode struct { flags uint32 // 是否为bucket pgid pgid // 从属的pgID key []byte // key value []byte // value(分支节点为空) } type node struct { bucket *Bucket // 从属bucket isLeaf bool // 是否为叶子节点 unbalanced bool // 是否平衡,用于reblance spilled bool // 是否执行过分裂 key []byte // 该节点最小key pgid pgid // 该node对应pgid parent *node // 父节点 children nodes // 子节点 inodes inodes // 存储键值对的结构体数组 }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
读写事务提交时，需要把所有
node序列化到page物理页，此时会调用node.write()方法进行序列化。// write writes the items onto one or more pages. func (n *node) write(p *page) { // 初始化page if n.isLeaf { p.flags = leafPageFlag } else { p.flags = branchPageFlag } // 记录子节点数量 p.count = uint16(len(n.inodes)) // 无数据时返回 if p.count == 0 { return } // 计算pageElement的偏移量 off := unsafe.Sizeof(*p) + n.pageElementSize()*uintptr(len(n.inodes)) for i, item := range n.inodes { // 计算key和value长度 sz := len(item.key) + len(item.value) // 计算当前key和value对应位置 b := unsafeByteSlice(unsafe.Pointer(p), off, 0, sz) // 更新偏移量 off += uintptr(sz) // 叶子节点 if n.isLeaf { // 获取第一个leaf elme元素 elem := p.leafPageElement(uint16(i)) // 更新key地址、长度及value长度 elem.pos = uint32(uintptr(unsafe.Pointer(&b[0])) - uintptr(unsafe.Pointer(elem))) elem.flags = item.flags elem.ksize = uint32(len(item.key)) elem.vsize = uint32(len(item.value)) } else { // 获取第一个branch elme元素 elem := p.branchPageElement(uint16(i)) // 更新key地址、长度及pgID elem.pos = uint32(uintptr(unsafe.Pointer(&b[0])) - uintptr(unsafe.Pointer(elem))) elem.ksize = uint32(len(item.key)) elem.pgid = item.pgid _assert(elem.pgid != p.id, "write: circular dependency occurred") } // 添加对应的key、value l := copy(b, item.key) // 分支节点此处不会生效 copy(b[l:], item.value) } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49读写事务运行过程中，涉及到
B+树的调整，会将page反序列化到node实例，内存中完成reblance。func (n *node) read(p *page) { // 记录节点类型、pgID n.pgid = p.id n.isLeaf = ((p.flags & leafPageFlag) != 0) n.inodes = make(inodes, int(p.count)) // 更新每个子节点信息 for i := 0; i < int(p.count); i++ { inode := &n.inodes[i] if n.isLeaf { // 解析当前leaf elem elem := p.leafPageElement(uint16(i)) // 更新子节点标识、key、value inode.flags = elem.flags inode.key = elem.key() inode.value = elem.value() } else { // 解析当前branch elme elem := p.branchPageElement(uint16(i)) // 更新key和pgID inode.pgid = elem.pgid inode.key = elem.key() } } // 记录起始key if len(n.inodes) > 0 { n.key = n.inodes[0].key } else { n.key = nil } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31key不断插入的过程中，某个节点上的元素数量会超出阈值，此时就会触发节点分裂，分裂的行为自下而上。事务的commit阶段，由root节点发起对有改动的子节点尝试分裂，递归向上调整，直至追溯到root。节点分裂是inodes列表分裂，通过累加inodes里的节点内存，size超出的位置就是分裂位置index。threshold := int(float64(pageSize) * fillPercent) // 分裂阈值 const minKeysPerPage = 2 // 每页最少key,分裂要求节点key数量大于该值2倍 // 节点分裂 func (n *node) split(pageSize uintptr) []*node { var nodes []*node node := n for { // 分裂2个节点 a, b := node.splitTwo(pageSize) nodes = append(nodes, a) // 新节点空的退出 if b == nil { break } // 更新node,下一次node会被再次用于分裂 node = b } return nodes } // 分裂两个节点 func (n *node) splitTwo(pageSize uintptr) (*node, *node) { // key太少不拆分 if len(n.inodes) <= (minKeysPerPage*2) || n.sizeLessThan(pageSize) { return n, nil } threshold := int(float64(pageSize) * fillPercent) // 计算分裂位置 splitIndex, _ := n.splitIndex(threshold) // 初始化父节点(bucket节点分裂) if n.parent == nil { n.parent = &node{bucket: n.bucket, children: []*node{n}} } // 初始化新分裂节点 next := &node{bucket: n.bucket, isLeaf: n.isLeaf, parent: n.parent} n.parent.children = append(n.parent.children, next) // 分裂 next.inodes = n.inodes[splitIndex:] n.inodes = n.inodes[:splitIndex] return n, next } // 计算分裂位置 func (n *node) splitIndex(threshold int) (index, sz uintptr) { sz = pageHeaderSize // 原节点保留最小key for i := 0; i < len(n.inodes)-minKeysPerPage; i++ { index = uintptr(i) inode := n.inodes[i] // 计算一个page elme及key、value占用长度 elsize := n.pageElementSize() + uintptr(len(inode.key)) + uintptr(len(inode.value)) // 原节点保留最小key时,找到分裂位置跳出 if index >= minKeysPerPage && sz+elsize > uintptr(threshold) { break } // 累计分裂阈值 sz += elsize } return }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
key删除时就涉及到节点合并，避免节点过多影响查询。合并是自上向下的，执行删除操作的node才会调整，同时子节点的调整触发parent节点尝试调整，当前节点调整后递归遍历子节点进行合并，合并也会根据阈值判断。1.unbalanced为false // 增删过 2.n.size() > threshold // 内存溢出 3.len(n.inodes) > n.minKeys() // 数据量超出页最小key数目 // 节点合并 func (n *node) rebalance() { // 节点平衡 if !n.unbalanced { return } n.unbalanced = false // 合并阈值 var threshold = n.bucket.tx.db.pageSize / 4 // size超出pageSize的25%且子元素数量大于最小值,不需要合并 if n.size() > threshold && len(n.inodes) > n.minKeys() { return } // 没有父节点(bucket节点合并) if n.parent == nil { // 分支节点且只有一个inode if !n.isLeaf && len(n.inodes) == 1 { // 加载子节点 child := n.bucket.node(n.inodes[0].pgid, n) // 子节点属性及inodes列表赋值给 n.isLeaf = child.isLeaf n.inodes = child.inodes[:] // 相当于子节点被删除,当前节点继承其子节点 n.children = child.children // 更新子节点的父亲 for _, inode := range n.inodes { if child, ok := n.bucket.nodes[inode.pgid]; ok { child.parent = n } } // 清理父子关系,释放child child.parent = nil delete(n.bucket.nodes, child.pgid) child.free() } return } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
func (n *node) rebalance() { ... // 非bucket、当前节点没有子节点合并 if n.numChildren() == 0 { // 移除当前节点,触发父节点合并 n.parent.del(n.key) n.parent.removeChild(n) delete(n.bucket.nodes, n.pgid) n.free() n.parent.rebalance() return } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 最左侧节点合并策略 func (n *node) rebalance(){ ... // 是否有上一个节点 var target *node var useNextSibling = (n.parent.childIndex(n) == 0) if useNextSibling { // 获取右兄弟 target = n.nextSibling() } else { // 获取左兄弟 target = n.prevSibling() } // 当前节点是parent最左侧节点 if useNextSibling { // 右兄弟子元素合并到当前节点 for _, inode := range target.inodes { if child, ok := n.bucket.nodes[inode.pgid]; ok { child.parent.removeChild(child) child.parent = n child.parent.children = append(child.parent.children, child) } } // 更新当前节点子元素,同时释放右兄弟节点 n.inodes = append(n.inodes, target.inodes...) n.parent.del(target.key) n.parent.removeChild(target) delete(n.bucket.nodes, target.pgid) target.free() } else { ... } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
// 其他情况合并 func (n *node) rebalance(){ ... if useNextSibling { ... // 其他情况,合并左兄弟节点 } else { // Reparent all child nodes being moved. for _, inode := range n.inodes { if child, ok := n.bucket.nodes[inode.pgid]; ok { child.parent.removeChild(child) child.parent = target child.parent.children = append(child.parent.children, child) } } // Copy over inodes to target and remove node. target.inodes = append(target.inodes, n.inodes...) n.parent.del(n.key) n.parent.removeChild(n) delete(n.bucket.nodes, n.pgid) n.free() } // 当前节点调整后触发父节点调整 n.parent.rebalance() }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 2.bucket
# 2.1.节点桶
bucket相当于命名空间，针对下层结构进一步组织抽象，bucket相当于一颗完整的B+树，用于联合多个节点实现快速检索。每个db默认就有一个bucket，新建的bucket都是其子节点，bucket键的嵌套组合可以组成B+树，当然其本身也是一颗B+树。type bucket struct { root pgid // bucket的根节点pgID,内联时为0 sequence uint64 // 单调递增的序列号,用于NextSequence()函数 } type Bucket struct { *bucket // tx *Tx // 指向关联事务的指针 buckets map[string]*Bucket // 子bucket缓存,用于通过名字快速访问 page *page // 内存页面的引用,用于直接存储少量数据或作为数据节点的入口 rootNode *node // 根页面已实例化的节点,bucket直接存储在内存此节点将被激活 nodes map[pgid]*node // 节点缓存,用于快速访问已加载的页面节点,避免重复读取磁盘 // 设置节点分裂时的填充阈值,默认情况下,bucket将填充至50% // 写入工作负载主要是追加操作时,可以提高这个比例 // 这个设置不会跨事务持久化,因此每个事务都必须设置它 FillPercent float64 }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 2.2.创建桶
CreateBucket根据指定的key创建bucket，bucket存在或key之前插入过元素会报错退出。否则，会在当前bucket中找到合适的位置，新建一个bucket写入并返回给客户端。// CreateBucket creates a new bucket at the given key and returns the new bucket. func (b *Bucket) CreateBucket(key []byte) (*Bucket, error) { ... // 初始化cursor c := b.Cursor() // 查找key(key不存在返回第一个比key大的数据) k, _, flags := c.seek(key) // key存在 if bytes.Equal(key, k) { // key所在节点是桶 if (flags & bucketLeafFlag) != 0 { return nil, ErrBucketExists } // key存在 return nil, ErrIncompatibleValue } // 初始化一个内联bucket var bucket = Bucket{ bucket: &bucket{}, rootNode: &node{isLeaf: true}, FillPercent: DefaultFillPercent, } // 写入bucketHeader var value = bucket.write() // 找到key所在叶子及写入 key = cloneBytes(key) c.node().put(key, key, value, 0, bucketLeafFlag) // 内联bucket中page=nil b.page = nil // 返回对应的bucket return b.Bucket(key), nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33bucket初始化会调用bucket.write()向bucket写入bucketHeader，调用c.node()找到key所在叶子节点，写入序列化后的bucket数据。为什么找node而不是bucket？这是因为真实的B+树存储的其实是一个个node，node会包装bucket。func (b *Bucket) write() []byte { // 当前bucket根节点 var n = b.rootNode // 创建一个可以容纳header+node数据的切片 var value = make([]byte, bucketHeaderSize+n.size()) // 深拷贝出bucketHeader内容 var bucket = (*bucket)(unsafe.Pointer(&value[0])) *bucket = *b.bucket // bucketHeader后的内容转换为page var p = (*page)(unsafe.Pointer(&value[bucketHeaderSize])) // rootNode序列化到page n.write(p) // 此时value=header+inline page return value }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
初建的
bucket是内联类型，并没有子节点，随着key增加，bucket会进行分裂，也就会展开为B+树。
关于新bucket写入
1.
seek()会从根节点开始层层二分查找key的位置2.
key满足某个节点范围，会定位到其叶子节点；否则，路径栈stack只会保留根节点，记录下一级节点索引0或size-13.
node()方法会找到叶子节点，put方法会将新建的bucket数据写入叶子节点的idx位置
# 2.3.检索桶
// 通过名称检索嵌套桶 // 返回的桶实例只有在事务周期有效 func (b *Bucket) Bucket(name []byte) *Bucket { // 已有缓存桶 if b.buckets != nil { if child := b.buckets[string(name)]; child != nil { return child } } // 初始化cursor c := b.Cursor() // 根据name查找bucket k, v, flags := c.seek(name) ... // 根据找到的value打开桶(反序列化为bucket) var child = b.openBucket(v) // 更新bucket缓存 if b.buckets != nil { b.buckets[string(name)] = child } // 返回找到的桶 return child } func (b *Bucket) openBucket(value []byte) *Bucket { // 初始化bucket var child = newBucket(b.tx) // 拷贝value数据 unaligned := uintptr(unsafe.Pointer(&value[0]))&unalignedMask != 0 if unaligned { value = cloneBytes(value) } // 存在写事务 if b.tx.writable && !unaligned { // 拷贝bucket值 child.bucket = &bucket{} *child.bucket = *(*bucket)(unsafe.Pointer(&value[0])) // 否则,直接指向内存映射 } else { child.bucket = (*bucket)(unsafe.Pointer(&value[0])) } // 内联桶 if child.root == 0 { // 内联桶的话直接读取页头后的数据 child.page = (*page)(unsafe.Pointer(&value[bucketHeaderSize])) } return &child }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# 2.4.删除桶
deleteBucket方法用来删除一个指定的Bucket，内部实现逻辑是先递归的删除其子桶，再释放该Bucket的page，最终从叶子节点中移除。func (b *Bucket) DeleteBucket(key []byte) error { ... // 初始化cursor c := b.Cursor() // 找到所在位置 k, _, flags := c.seek(key) ... // 获取对应bucket child := b.Bucket(key) // 递归删除每个子桶 err := child.ForEachBucket(func(k []byte) error { if err := child.DeleteBucket(k); err != nil { return fmt.Errorf("delete bucket: %s", err) } return nil }) if err != nil { return err } // 删除bucket缓存 delete(b.buckets, string(key)) // 释放页节点,加入空闲页列表 child.nodes = nil child.rootNode = nil child.free() // 删除key c.node().del(key) return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 2.5.合并
前面介绍了
node副本的rebalance，其实rebalance的入口是bucket。所有反序列化的node副本会执行rebalance，之后就是对所有反序列化过的bucket副本执行rebalance。func (b *Bucket) rebalance() { // 1.node副本未删除过或已经执行rebalance,结束 // 2.标记当前node执行过rebalance // 3.nodeSize>pageSize/4且len(inodes)>minKeys,无需合并 // 4.node副本是root且只有一个子节点,子节点合并到node作为inline // 5.node副本无子节点,移除node结束 // 6.node副本是父节点最左侧,与右兄弟节点合并;否则和左兄弟节点合并 for _, n := range b.nodes { n.rebalance() } // 涉及反序列化的bucket调整 for _, child := range b.buckets { child.rebalance() } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 2.6.分裂
前文已经介绍过
node副本的分裂，其实spill的入口是bucket。rebalance完成后，会进一步执行spill操作，spill会完成大节点的拆分，也会基于copy on write机制为所有反序列化的node分配新的page副本。合并后的再分裂，其实也是避免小节点合并为大节点，超出size及len(inodes)限制的手段。// 1.针对所有反序列化的bucket执行分裂 // 2.分裂后生成新的序列化内容写入当前bucket副本的B+树 // 3.沿着bucket root依次以node副本的维度执行分裂 func (b *Bucket) spill() error { // 所有反序列化bucket执行分裂 for name, child := range b.buckets { // 内联类型无需分裂 var value []byte if child.inlineable() { child.free() value = child.write() // 非内联类型执行分裂 } else { if err := child.spill(); err != nil { return err } // 深拷贝bucketHeader value = make([]byte, unsafe.Sizeof(bucket{})) var bucket = (*bucket)(unsafe.Pointer(&value[0])) *bucket = *child.bucket } // child bucket无数据 if child.rootNode == nil { continue } // 初始化cursor var c = b.Cursor() // 查找bucket k, _, flags := c.seek([]byte(name)) ... // 取叶子节点更新bucket,value是分裂后新的bucket内容 c.node().put([]byte(name), []byte(name), value, 0, bucketLeafFlag) } // bucket无数据 if b.rootNode == nil { return nil } // bucket沿着root分裂 // 1.node副本已执行过分裂,结束 // 2.反序列化的子节点分裂 // 3.node副本执行分裂 // 4.每个node副本申请新的page副本,体现copy on write机制 if err := b.rootNode.spill(); err != nil { return err } // 更新root引用(避免分裂后出现新的root) b.rootNode = b.rootNode.root() ... // 更新root节点pgID b.root = b.rootNode.pgid return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# 3.cursor
# 3.1.mmap
boltdb采用内存映射的方式将整个db文件映射到进程的虚拟内存空间，访问某页数据时才会真正读取物理空间的page初始化为node，继而进行inodes新增、更新或删除，接着内存映射方式直接写入。boltdb采用的mmap内存映射技术减少了read时内核缓冲区拷贝，将数据直接拷贝到物理内存，进程虚拟地址指向对应物理内存。
- 运行过程中，不会因为单笔改动频繁执行
rebalance和spill操作，事务提交时一次性完成B+树的自平衡调整，完整的内容会重新持久化为稳定版本的磁盘B+树。 
- 树的根是一个默认
bucket，第一层的bucket可以看作分支节点。节点在内存中是node对象，子元素就是node对象的inodes列表数据，node属于叶子时inode就是key/value，否则子元素只是用于索引，保证一个页节点尽可能多方索引key，达到最少的IO索引目标。 
# 3.2.索引
cursor用于快速遍历B+树，查找bucket中key对应的位置，一个bucket对象关联一个cursor，通过cursor按照中序遍历方式访问树节点，cursor.stack会存储中序遍历路径。type elemRef struct { page *page // 对应page(加载到内存中) node *node // node(page未加载到内存) index int // 路径经过该节点时在inodes中的位置 } type Cursor struct { bucket *Bucket // 进行node加载 stack []elemRef // 保留路径,方便回溯 }1
2
3
4
5
6
7
8
9
10
# 3.3.seek
Seek()用于快速查找≥key的最接近元素，实现上会将≤key的最接近元素记录到stack，没找到key情况下从栈顶回退到父节点，找到next元素作为返回值。// 查询≥key的最接近元素 func (c *Cursor) Seek(seek []byte) (key []byte, value []byte) { ... // seek查找 k, v, flags := c.seek(seek) // 栈顶节点遍历后没找到,尝试回退到父节点找下一个 if ref := &c.stack[len(c.stack)-1]; ref.index >= ref.count() { k, v, flags = c.next() } if k == nil { return nil, nil // 嵌套bucket } else if (flags & uint32(bucketLeafFlag)) != 0 { // 返回key return k, nil } return k, v } // seek搜索 func (c *Cursor) seek(seek []byte) (key []byte, value []byte, flags uint32) { // 清空路径栈 c.stack = c.stack[:0] // bucket的根节点开始找 c.search(seek, c.bucket.root) // 解析key,value,flag return c.keyValue() }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30seek查询时会调用search方式，沿着pgID作为起点出发，将cursor移动到B+树中≤key的最接近位置，同时将层层路径压入stack记录。// 1.获取pgID对应节点,已反序列化使用node,否则使用page // 2.节点包装为elmeRef,压入cursor的stack // 3.叶子节点,通过nsearch在节点内查找数据 // 4.分支节点,利用已经反序列化的node或page向下检索 func (c *Cursor) search(key []byte, pgId pgid) { // 根据pgid读出当前节点,未反序列化node就加载页 p, n := c.bucket.pageNode(pgId) ... // 包装移动路径入栈 e := elemRef{page: p, node: n} c.stack = append(c.stack, e) // 当前节点是叶子,节点内部检索结果 if e.isLeaf() { c.nsearch(key) return } // 分支节点向下检索 if n != nil { c.searchNode(key, n) return } // 分支页向下检索 c.searchPage(key, p) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26nsearch用于节点内检索数据，由于内存B+树遵循懒加载机制，因此访问节点时会先用node，未反序列化时才会用page加载数据。// 1.获取栈顶叶子节点 // 2.二分查找首个≥key的位置 // 3.记录最佳位置,用于后续计算 func (c *Cursor) nsearch(key []byte) { // 拿到当前页末尾节点 e := &c.stack[len(c.stack)-1] p, n := e.page, e.node // 加载到node,从node二分查找满足key的数据 if n != nil { index := sort.Search(len(n.inodes), func(i int) bool { return bytes.Compare(n.inodes[i].key, key) != -1 }) e.index = index return } // 未加载到node,从页中查找 inodes := p.leafPageElements() // 二分查找 index := sort.Search(int(p.count), func(i int) bool { return bytes.Compare(inodes[i].key(), key) != -1 }) // 记录满足key的数据位置 e.index = index }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24searchNode和searchPage用于递归检索分支节点，检索原理相同，基于二分找到最佳位置后，记录≤key的最佳位置，用于key未找到时回退至父节点的next。// 1.二分找到≥key的首个位置 // 2.找到key,记录key的位置;否则记录≤key的最接近位置 // 3.找到的目标作为新起点,调用search继续检索 func (c *Cursor) searchNode(key []byte, n *node) { var exact bool // 节点内二分查找 index := sort.Search(len(n.inodes), func(i int) bool { ret := bytes.Compare(n.inodes[i].key, key) if ret == 0 { exact = true } return ret != -1 }) // 记录≤key最接近位置 if !exact && index > 0 { index-- } c.stack[len(c.stack)-1].index = index // 递归到子节点 c.search(key, n.inodes[index].pgid) } func (c *Cursor) searchPage(key []byte, p *common.Page) { // 分支页的子节点 inodes := p.BranchPageElements() var exact bool // 二分查找最佳位置 index := sort.Search(int(p.Count()), func(i int) bool { ret := bytes.Compare(inodes[i].Key(), key) if ret == 0 { exact = true } return ret != -1 }) // 记录≤key最接近位置 if !exact && index > 0 { index-- } c.stack[len(c.stack)-1].index = index // 递归搜索到下一页 c.search(key, inodes[index].Pgid()) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# 3.4.node
node用于返回游标移动路径中的最后一个叶子节点的node副本，stack顶部是叶子节点时会直接返回，否则向栈底移动找到最后一个叶子节点。func (c *Cursor) node() *node { // 栈顶是叶子节点 if ref := &c.stack[len(c.stack)-1]; ref.node != nil && ref.isLeaf() { return ref.node } // 根节点兜底 var n = c.stack[0].node if n == nil { n = c.bucket.node(c.stack[0].page.id, nil) } // 扣除栈顶元素,根节点出发向栈顶遍历 for _, ref := range c.stack[:len(c.stack)-1] { // 按照移动路径向下 n = n.childAt(ref.index) } return n } // 返回移动路径中的节点副本(兼容inline bucket) func (n *node) childAt(index int) *node { ... return n.bucket.node(n.inodes[index].pgid, n) } func (b *Bucket) node(pgId pgid, parent *node) *node { // 反序列化过node,返回复用 if n := b.nodes[pgId]; n != nil { return n } // 构造node副本实例 n := &node{bucket: b, parent: parent} if parent == nil { b.rootNode = n } else { parent.children = append(parent.children, n) } // 非inline bucket,取page var p = b.page if p == nil { p = b.tx.page(pgId) } // page反序列化到node n.read(p) // 缓存node b.nodes[pgId] = n return n }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# 3.5.put
bucket.put可以向指定表中插入或更新一组KV数据，实现时依赖游标查询key所在叶子节点，未找到时创建KV，找到则更新。这里不涉及创建新的node，只有写入inode数据溢出，当前节点才会分裂。// 1.初始化一个游标 // 2.借助cursor.seek查询key // 3.借助node.put向节点最佳位置写入KV func (b *Bucket) Put(key []byte, value []byte) error { ... // 初始化游标 c := b.Cursor() // 查询key最佳位置 k, _, flags := c.seek(key) // 找到key且属于sub bucket if bytes.Equal(key, k) && (flags&bucketLeafFlag) != 0 { return ErrIncompatibleValue } // 获取key所在叶子节点,写入或更新key key = cloneBytes(key) c.node().put(key, key, value, 0, 0) return nil } func (n *node) put(oldKey, newKey, value []byte, pgId pgid, flags uint32) { ... // 二分查找key index := sort.Search(len(n.inodes), func(i int) bool { return bytes.Compare(n.inodes[i].key, oldKey) != -1 }) // key是否存在 exact := (len(n.inodes) > 0 && index < len(n.inodes) && bytes.Equal(n.inodes[index].key, oldKey)) // 不存在,插入到index if !exact { // 将index后的元素右移一位 n.inodes = append(n.inodes, inode{}) copy(n.inodes[index+1:], n.inodes[index:]) } // 更新选中节点 inode := &n.inodes[index] inode.flags = flags inode.key = newKey inode.value = value inode.pgid = pgId }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# 3.6.get
bucket.get()用于从表中获取KV，其实也是借助cursor找到最佳路径，利用cursor.stack栈顶的叶子节点获取KV数据。// 1.初始化cursor,基于cursor.seek()查找最佳路径 // 2.取出cursor.stack叶子节点中≥key最小值 // 3.根据key的值返回value func (b *Bucket) Get(key []byte) []byte { // 初始化游标查询key k, v, flags := b.Cursor().seek(key) // 查询到的key在sub bucket if (flags & bucketLeafFlag) != 0 { return nil } // key未找到 if !bytes.Equal(key, k) { return nil } return v }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 3.5.delete
bucket.delete()用于实现表中KV删除，借助cursor实例找到key所在叶子节点，从node.inodes列表删除key。// 1.初始化cursor // 2.移动cursor,找到key的路径 // 3.取出cursor.stack栈顶的叶子节点,检查key是否存在 // 4.存在时,基于node.del移除key对应inode func (b *Bucket) Delete(key []byte) error { ... // 初始化cursor c := b.Cursor() // 查找key的最佳路径 k, _, flags := c.seek(key) // key不存在 if !bytes.Equal(key, k) { return nil } // 找到key所在叶子节点,删除key对应inode c.node().del(key) return nil } func (n *node) del(key []byte) { // 二分找到≥key最小值 index := sort.Search(len(n.inodes), func(i int) bool { return bytes.Compare(n.inodes[i].key, key) != -1 }) // key不存在 if index >= len(n.inodes) || !bytes.Equal(n.inodes[index].key, key) { return } // 删除key对应inode n.inodes = append(n.inodes[:index], n.inodes[index+1:]...) // 标记节点再平衡 n.unbalanced = true }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# 4.事务
# 4.1.定义
boltdb支持ACID事务采用读写锁机制支持多读与单写并发执行，每个事务都会关联一个txid。读事务执行时，将meta.txid作为txid，写事务执行时，meta.txid+1作为txid，写事务成功提交后，更新meta.txid为写事务的txid。type txid uint64 // 1.没有事务使用页面时,写入者才能回收这些页面 // 2.长时间运行的读事务可能导致数据库迅速增加 type Tx struct { writable bool // 是否为可写事务 managed bool // 是否为管理事务 db *DB // 从属的DB meta *meta // 元数据页 root Bucket // 根桶 pages map[pgid]*page // 页面映射 commitHandlers []func() // 事务提交的回调 WriteFlag int // 写标志(syscall.O_DIRECT) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
DB中也定义了事务相关部分，包括两个meta page、读写事务rwtx、只读事务txs以及事务锁rwlock。type DB struct { ... meta0 *meta // meta page,相当于一个版本的数据源头 meta1 *meta // 轮换的meta page,相当于一个版本的数据源头 ... rwtx *Tx // 全局唯一的读写事务 txs []*Tx // 一系列并行的只读事务 ... rwlock sync.Mutex // 事务锁 ... }1
2
3
4
5
6
7
8
9
10
11
# 4.2.原子性
boltdb采用写时复制保证事务的原子性，所有产生中间态的数据临时缓存到内存副本，事务提交时变更内容一次性持久化。针对事务初始化，基于最新版本的meta page和关联的bucket深拷贝一份副本。func (tx *Tx) init(db *DB) { ... // 基于copy-on-write拷贝meta page tx.meta = &meta{} db.meta().copy(tx.meta) // 基于copy-on-write拷贝bucket副本,基于meta page持有其引用 tx.root = newBucket(tx) tx.root.bucket = &bucket{} *tx.root.bucket = tx.meta.root // 针对写事务,更新txid if tx.writable { tx.pages = make(map[pgid]*page) tx.meta.txid += txid(1) } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17事务执行过程中，数据修改通过
page反序列化出对应的内存node，基于node副本进行修改。func (b *Bucket) node(pgId pgid, parent *node) *node { ... // 已反序列化node if n := b.nodes[pgId]; n != nil { return n } // 构造node副本 n := &node{bucket: b, parent: parent} ... // 根据pgID从mmap page buffer读取page var p = b.page if p == nil { p = b.tx.page(pgId) } // page反序列化为node n.read(p) // 更新node缓存 b.nodes[pgId] = n ... return n }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23事务提交时，所有的
node副本序列化为新的page——dirty page，同时将其嵌入meta page副本下的B+树副本，不会直接影响原page中上一个版本的数据。func (tx *Tx) Commit() error { ... // 根bucket开始合并 tx.root.rebalance() ... opgid := tx.meta.pgid ... // 根bucket开始分裂 if err := tx.root.spill(); err != nil { tx.rollback() return err } // 更新meta page的bucket tx.meta.root.root = tx.root.root // 释放空闲列表已申请的页 if tx.meta.freelist != pgidNoFreelist { tx.db.freelist.free(tx.meta.txid, tx.db.page(tx.meta.freelist)) } // 最新的空闲列表序列化到连续的可用新页,更新meta.freelist.pgID if !tx.db.NoFreelistSync { // 每次提交时,会把freelist分配到可用的连续新页 // 更新meta.freelist记录pgID,这样就保证每次落盘,meta关联的freelist都是最新的 err := tx.commitFreelist() ... } // pgID更新,计算dataSz是否溢出dbFile if tx.meta.pgid > opgid { // dbFile不够落盘,扩容申请空间 // 1.db.dataSz <= dbFile,不需要申请 // 2.db.dataSz <= db.AllocSize,则sz = db.dataSz // 3.db.dataSz > db.AllocSize,则sz += db.AllocSize if err := tx.db.grow(int(tx.meta.pgid+1) * tx.db.pageSize); err != nil { tx.rollback() return err } } // dirty page溢写落盘 if err := tx.write(); err != nil { tx.rollback() return err } ... // meta page溢写落盘(为什么只写meta0,后台备份到meta1吗) if err := tx.writeMeta(); err != nil { tx.rollback() return err } ... // 事务结束 tx.close() ... return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
注意
1.meta page记录了B+树的入口，因为数据是否生效由meta page落盘结果决定
2.freelist和dirty page落盘不代表数据生成，meta落盘后数据才会正式生效
# 4.3.隔离性
boltdb事务隔离级别介于可重复读和串行化之间，全局只有一个的读写事务执行模式上串行化，并行执行的读事务满足可重复读的语义。实现上，读写事务启动时通过写锁互斥，保证数据唯一性。func (db *DB) beginRWTx() (*Tx, error) { ... db.rwlock.Lock() ... t := &Tx{writable: true} t.init(db) db.rwtx = t db.freePages() return t, nil }1
2
3
4
5
6
7
8
9
10只读事务运行时，不会取事务锁，直接追加到只读事务列表。
func (tx *Tx) close() { ... if tx.writable { ... // 读写事务解锁 tx.db.rwlock.Unlock() ... } else { // 移除读事务 tx.db.removeTx(tx) } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
可重复读的核心是copy-on-write和pending机制，通过拷贝上一版本的meta page副本作为新的入口，事务提交时会将上一版本的数据置为pending，上一版本的page没有读事务时才会从pending释放。// 追加page到pending page list func (f *freelist) free(txid txid, p *page) { ... // 获取对应事务pending page list txp := f.pending[txid] ... // free的page追加到pending page list(所有溢出页) for id := p.id; id <= p.id+pgid(p.overflow); id++ { ... txp.ids = append(txp.ids, id) ... } } // 所有 ≤ txid的pending page进行释放 func (f *freelist) release(txid txid) { m := make(pgids, 0) for tid, txp := range f.pending { if tid <= txid { // pending page追加到release list m = append(m, txp.ids...) // 清理txid的pending page list delete(f.pending, tid) } } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
事务初始化时，基于当前
txid最大的meta page深拷贝出一份副本，该副本作为整个事务运行期间的数据版本，对于读事务来说该副本不会改变，因此可以保证可重复读的语义。func (tx *Tx) init(db *DB) { ... // 拷贝meta副本 tx.meta = &meta{} db.meta().copy(tx.meta) // 拷贝bucket副本 tx.root = newBucket(tx) tx.root.bucket = &bucket{} *tx.root.bucket = tx.meta.root }1
2
3
4
5
6
7
8
9
10
事务执行过程中，修改数据会将
page反序列化为node于内存中修改，通过事务提交使副本生效。这个过程不会对原本的page产生影响，基于新建的脏数据page副本承载数据，对于原本使用的旧page不会立即释放。free page的调用时机发生在事务提交流程，此时会涉及node的spill流程，所有改动的node副本的旧page添加到当前事务的pending page list中。
每当有新的读写事务启动，会记录当前仍然运行的最小
txid，同时将小于该值的事务对应的pending page list全部release。func (db *DB) beginRWTx() (*Tx, error) { ... db.freePages() return t, nil } func (db *DB) freePages() { // 基于txid对pending page list排序 sort.Sort(txsById(db.txs)) // 获取仍在运行的最小只读事务txid minid := txid(0xFFFFFFFFFFFFFFFF) if len(db.txs) > 0 { minid = db.txs[0].meta.txid } // 小于txid的pending page list释放 if minid > 0 { db.freelist.release(minid - 1) } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
注意
1.
freelist会关联到DB，也就是运行期间freelist复用2.事务提交时会把
used page记录到db.freelist的pending page list3.每开启新事务，释放小于
mini txid的pending page list4.
meta page中的freelist用于初始化db时加载freelist，实际不起任何作用
# 4.4.持久性
事务的持久性基于
pwrite + fdatasync操作，pwrite操作将dirty page溢写到文件的指定offset，通过fdatasync确保该文件下的设备io操作完成。// 提交事务 func (tx *Tx) Commit() error { ... // 基于pwrite + fdatasync操作实现脏数据page持久化落盘 if err := tx.write(); err != nil { ... } ... // 基于pwrite + fdatasync操作实现meta page副本持久化落盘 // 基于txid%2决定写入哪个meta,也就是每次轮流写入 if err := tx.writeMeta(); err != nil { ... } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15meta page副本成功落盘，代表读写数据成功提交，新的数据版本正式生效。meta page落盘失败时，事务启动时读取的meta page会因为校验和检查失败被视为失效版本，此时会基于冗余的另一份meta page正常对外提供服务。// 读取数据中txid更新且有效的meta page func (db *DB) meta() *meta { // 优先获取txid更大的meta page metaA := db.meta0 metaB := db.meta1 if db.meta1.txid > db.meta0.txid { metaA = db.meta1 metaB = db.meta0 } // 拷贝有效的版本 if err := metaA.validate(); err == nil { return metaA } else if err := metaB.validate(); err == nil { return metaB } ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 4.5.mmap
mmap全称memory mapping，用于实现磁盘文件与内存空间的映射，屏蔽磁盘IO细节，便于访问内存的字节数组一样操作磁盘文件内容。boltdb中mmap的大小上限为256TB，内存映射后对应内容通过db.data字段获取，利用读写锁mmaplock保证并发安全。const maxMapSize = 0xFFFFFFFFFFFF // 256TB type DB struct { ... path string // db文件路径 openFile func(string, int, os.FileMode) (*os.File, error) // 打开db文件方法 file *os.File // db文件 ... dataref []byte // 原始字节数组,只读 data *[maxMapSize]byte // mmap映射db文件得到的字节数组 datasz int // mmap size ... mmaplock sync.RWMutex // 保护mmap操作并发安全的锁 ... }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15db启动或事务提交时，page不足会执行mmap操作，前者用于初始化，后者用于扩容。// 启动boltdb func Open(path string, mode os.FileMode, options *Options) (*DB, error) { // 1.构造db实例 ... // 2.读取配置 ... // 3.打开db文件 ... // 4.初始化db文件前4个page ... // 5.执行mmap操作 if err := db.mmap(options.InitialMmapSize); err != nil { _ = db.close() return nil, err } ... // 6.返回db实例 return db, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19事务提交或
page反序列化为node副本操作时，调用allocate申请指定数量page。// 申请count页 func (db *DB) allocate(txid txid, count int) (*page, error) { ... // 1.尝试从freelist申请 if p.id = db.freelist.allocate(txid, count); p.id != 0 { return p, nil } // 2.mmap扩容 p.id = db.rwtx.meta.pgid var minsz = int((p.id+pgid(count))+1) * db.pageSize // 申请count页超出mmap size,申请 if minsz >= db.datasz { if err := db.mmap(minsz); err != nil { return nil, fmt.Errorf("mmap allocate error: %s", err) } } ... return p, nil } // 内存映射 func (db *DB) mmap(minsz int) error { // 1.加锁 db.mmaplock.Lock() defer db.mmaplock.Unlock() ... // 2.调整合适的mmap size size, err = db.mmapSize(size) ... // 3.读写事务运行时,更新其引用的B+树节点 // 所有B+树节点反序列化到内存,用于持久化时更新page信息 if db.rwtx != nil { db.rwtx.root.dereference() } // 4.已进行过mmap,需要解除映射 if err := db.munmap(); err != nil { return err } // 5.建立新的mmap映射 if err := mmap(db, size); err != nil { return err } ... // 6.校验meta page合法性 db.meta0 = db.page(0).meta() db.meta1 = db.page(1).meta() err0 := db.meta0.validate() err1 := db.meta1.validate() if err0 != nil && err1 != nil { return err0 } return nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53由于
mmap系统调用的成本较高，避免执行太频繁，boltdb每轮执行mmap时预留一部分空间：新容量小于1GB时，从32KB基础上持续倍增找到满足条件的容量，否则基于GB向上取整。const maxMmapStep = 1 << 30 // 1GB // 获取合适的mmap大小 func (db *DB) mmapSize(size int) (int, error) { // 32KB-1GB 范围内采取倍增策略 for i := uint(15); i <= 30; i++ { if size <= 1<<i { return 1 << i, nil } } ... // 大于1GB,向上GB取整 sz := int64(size) if remainder := sz % int64(maxMmapStep); remainder > 0 { sz += int64(maxMmapStep) - remainder } ... return int(sz), nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
