第二节 Bucket遍历之Cursor

第二节 Bucket遍历之Cursor

本节我们先做一节内容的铺垫，暂时不讲如何创建、获取、删除一个Bucket。而是介绍一个boltdb中的新对象Cursor。

答案是：所有的上述操作都是建立在首先定位到一个Bucket所属的位置，然后才能对其进行操作。而定位一个Bucket的功能就是由Cursor来完成的。所以我们先这一节给大家介绍一下boltdb中的Cursor。

我们先看下官方文档对Cursor的描述

Cursor represents an iterator that can traverse over all key/value pairs in a bucket in sorted order.

用大白话讲，既然一个Bucket逻辑上是一颗b+树，那就意味着我们可以对其进行遍历。前面提到的set、get操作，无非是要在Bucket上先找到合适的位置，然后再进行操作。而“找”这个操作就是交由Cursor来完成的。简而言之对Bucket这颗b+树的遍历工作由Cursor来执行。一个Bucket对象关联一个Cursor。下面我们先看看Bucket和Cursor之间的关系。

1. // Cursor creates a cursor associated with the bucket.
2. // The cursor is only valid as long as the transaction is open.
3. // Do not use a cursor after the transaction is closed.
4. func (b *Bucket) Cursor() *Cursor {
5. // Update transaction statistics.
6. b.tx.stats.CursorCount++

8. // Allocate and return a cursor.
9. return &Cursor{
10. bucket: b,
11. stack:  make([]elemRef, 0),
12. }
13. }

3.2.1 Cursor结构

从上面可以清楚的看到，在获取一个游标Cursor对象时，会将当前的Bucket对象传进去，同时还初始化了一个栈对象，结合数据结构中学习的树的知识。我们也就知道，它的内部就是对树进行遍历。下面我们详细介绍Cursor这个人物。

1. // Cursor represents an iterator that can traverse over all key/value pairs in a bucket in sorted order.
2. // Cursors see nested buckets with value == nil.
3. // Cursors can be obtained from a transaction and are valid as long as the transaction is open.
4. //
5. // Keys and values returned from the cursor are only valid for the life of the transaction.
6. //
7. // Changing data while traversing with a cursor may cause it to be invalidated
8. // and return unexpected keys and/or values. You must reposition your cursor
9. // after mutating data.
10. type Cursor struct {
11. bucket *Bucket
12. // 保存遍历搜索的路径
13. stack []elemRef
14. }

16. // elemRef represents a reference to an element on a given page/node.
17. type elemRef struct {
18. page  *page
19. node  *node
20. index int
21. }

23. // isLeaf returns whether the ref is pointing at a leaf page/node.
24. func (r *elemRef) isLeaf() bool {
25. if r.node != nil {
26. return r.node.isLeaf
27. }
28. return (r.page.flags & leafPageFlag) != 0
29. }

31. // count returns the number of inodes or page elements.
32. func (r *elemRef) count() int {
33. if r.node != nil {
34. return len(r.node.inodes)
35. }
36. return int(r.page.count)
37. }

3.2.2 Cursor对外接口

下面我们看一下Cursor对外暴露的接口有哪些。看之前也可以心里先想一下。针对一棵树我们需要哪些遍历接口呢？

主体也就是三类：定位到某一个元素的位置、在当前位置从前往后找、在当前位置从后往前找。

1. // First moves the cursor to the first item in the bucket and returns its key and value.
2. // If the bucket is empty then a nil key and value are returned.
3. // The returned key and value are only valid for the life of the transaction.
4. func (c *Cursor) First() (key []byte, value []byte)

6. // Last moves the cursor to the last item in the bucket and returns its key and value.
7. // If the bucket is empty then a nil key and value are returned.
8. // The returned key and value are only valid for the life of the transaction.
9. func (c *Cursor) Last() (key []byte, value []byte)

11. // Next moves the cursor to the next item in the bucket and returns its key and value.
12. // If the cursor is at the end of the bucket then a nil key and value are returned.
13. // The returned key and value are only valid for the life of the transaction.
14. func (c *Cursor) Next() (key []byte, value []byte)

16. // Prev moves the cursor to the previous item in the bucket and returns its key and value.
17. // If the cursor is at the beginning of the bucket then a nil key and value are returned.
18. // The returned key and value are only valid for the life of the transaction.
19. func (c *Cursor) Prev() (key []byte, value []byte)

21. // Delete removes the current key/value under the cursor from the bucket.
22. // Delete fails if current key/value is a bucket or if the transaction is not writable.
23. func (c *Cursor) Delete() error

25. // Seek moves the cursor to a given key and returns it.
26. // If the key does not exist then the next key is used. If no keys
27. // follow, a nil key is returned.
28. // The returned key and value are only valid for the life of the transaction.
29. func (c *Cursor) Seek(seek []byte) (key []byte, value []byte)

下面我们详细分析一下Seek()、First()、Last()、Next()、Prev()、Delete()这三个方法的内部实现。其余的方法我们代码就不贴出来了。大致思路可以梳理一下。

3.2.3 Seek(key)实现分析

Seek()方法内部主要调用了seek()私有方法，我们重点关注seek()这个方法的实现，该方法有三个返回值，前两个为key、value、第三个为叶子节点的类型。前面提到在boltdb中，叶子节点元素有两种类型：一种是嵌套的子桶、一种是普通的key/value。而这二者就是通过flags来区分的。如果叶子节点元素为嵌套的子桶时，返回的flags为1，也就是bucketLeafFlag取值。

1. // Seek moves the cursor to a given key and returns it.
2. // If the key does not exist then the next key is used. If no keys
3. // follow, a nil key is returned.
4. // The returned key and value are only valid for the life of the transaction.
5. func (c *Cursor) Seek(seek []byte) (key []byte, value []byte) {
6. k, v, flags := c.seek(seek)

8. // If we ended up after the last element of a page then move to the next one.
9. // 下面这一段逻辑是必须的，因为在seek()方法中，如果ref.index>ref.count()的话，就直接返回nil,nil,0了
10. // 这里需要返回下一个
11. if ref := &c.stack[len(c.stack)-1]; ref.index >= ref.count() {
12. k, v, flags = c.next()
13. }

15. if k == nil {
16. return nil, nil
17. //     子桶的话
18. } else if (flags & uint32(bucketLeafFlag)) != 0 {
19. return k, nil
20. }
21. return k, v
22. }

24. // seek moves the cursor to a given key and returns it.
25. // If the key does not exist then the next key is used.
26. func (c *Cursor) seek(seek []byte) (key []byte, value []byte, flags uint32) {
27. _assert(c.bucket.tx.db != nil, "tx closed")

29. // Start from root page/node and traverse to correct page.
30. c.stack = c.stack[:0]
31. // 开始根据seek的key值搜索root
32. c.search(seek, c.bucket.root)
33. // 执行完搜索后，stack中保存了所遍历的路径
34. ref := &c.stack[len(c.stack)-1]

36. // If the cursor is pointing to the end of page/node then return nil.
37. if ref.index >= ref.count() {
38. return nil, nil, 0
39. }
40. //获取值
41. // If this is a bucket then return a nil value.
42. return c.keyValue()
43. }

47. // keyValue returns the key and value of the current leaf element.
48. func (c *Cursor) keyValue() ([]byte, []byte, uint32) {
49. //最后一个节点为叶子节点
50. ref := &c.stack[len(c.stack)-1]
51. if ref.count() == 0 || ref.index >= ref.count() {
52. return nil, nil, 0
53. }

55. // Retrieve value from node.
56. // 先从内存中取
57. if ref.node != nil {
58. inode := &ref.node.inodes[ref.index]
59. return inode.key, inode.value, inode.flags
60. }

62. // 其次再从文件page中取
63. // Or retrieve value from page.
64. elem := ref.page.leafPageElement(uint16(ref.index))
65. return elem.key(), elem.value(), elem.flags
66. }

seek()中最核心的方法就是调用search()了，search()方法中，传入的就是要搜索的key值和该桶的root节点。search()方法中，其内部是通过递归的层层往下搜索，下面我们详细了解一下search()内部的实现机制。

1. // 从根节点开始遍历
2. // search recursively performs a binary search against a given page/node until it finds a given key.
3. func (c *Cursor) search(key []byte, pgid pgid) {
4. // root，3
5. // 层层找page，bucket->tx->db->dataref
6. p, n := c.bucket.pageNode(pgid)
7. if p != nil && (p.flags&(branchPageFlag|leafPageFlag)) == 0 {
8. panic(fmt.Sprintf("invalid page type: %d: %x", p.id, p.flags))
9. }
10. e := elemRef{page: p, node: n}
11. //记录遍历过的路径
12. c.stack = append(c.stack, e)

14. // If we're on a leaf page/node then find the specific node.
15. // 如果是叶子节点，找具体的值node
16. if e.isLeaf() {
17. c.nsearch(key)
18. return
19. }

21. if n != nil {
22. // 先搜索node，因为node是加载到内存中的
23. c.searchNode(key, n)
24. return
25. }
26. // 其次再在page中搜索
27. c.searchPage(key, p)
28. }

30. // pageNode returns the in-memory node, if it exists.
31. // Otherwise returns the underlying page.
32. func (b *Bucket) pageNode(id pgid) (*page, *node) {
33. // Inline buckets have a fake page embedded in their value so treat them
34. // differently. We'll return the rootNode (if available) or the fake page.
35. // 内联页的话，就直接返回其page了
36. if b.root == 0 {
37. if id != 0 {
38. panic(fmt.Sprintf("inline bucket non-zero page access(2): %d != 0", id))
39. }
40. if b.rootNode != nil {
41. return nil, b.rootNode
42. }
43. return b.page, nil
44. }

46. // Check the node cache for non-inline buckets.
47. if b.nodes != nil {
48. if n := b.nodes[id]; n != nil {
49. return nil, n
50. }
51. }

53. // Finally lookup the page from the transaction if no node is materialized.
54. return b.tx.page(id), nil
55. }

58. //node中搜索
59. func (c *Cursor) searchNode(key []byte, n *node) {
60. var exact bool
61. //二分搜索
62. index := sort.Search(len(n.inodes), func(i int) bool {
63. // TODO(benbjohnson): Optimize this range search. It's a bit hacky right now.
64. // sort.Search() finds the lowest index where f() != -1 but we need the highest index.
65. ret := bytes.Compare(n.inodes[i].key, key)
66. if ret == 0 {
67. exact = true
68. }
69. return ret != -1
70. })
71. if !exact && index > 0 {
72. index--
73. }
74. c.stack[len(c.stack)-1].index = index

76. // Recursively search to the next page.
77. c.search(key, n.inodes[index].pgid)
78. }

80. //页中搜索
81. func (c *Cursor) searchPage(key []byte, p *page) {
82. // Binary search for the correct range.
83. inodes := p.branchPageElements()

85. var exact bool
86. index := sort.Search(int(p.count), func(i int) bool {
87. // TODO(benbjohnson): Optimize this range search. It's a bit hacky right now.
88. // sort.Search() finds the lowest index where f() != -1 but we need the highest index.
89. ret := bytes.Compare(inodes[i].key(), key)
90. if ret == 0 {
91. exact = true
92. }
93. return ret != -1
94. })
95. if !exact && index > 0 {
96. index--
97. }
98. c.stack[len(c.stack)-1].index = index

100. // Recursively search to the next page.
101. c.search(key, inodes[index].pgid)
102. }

105. // nsearch searches the leaf node on the top of the stack for a key.
106. // 搜索叶子页
107. func (c *Cursor) nsearch(key []byte) {
108. e := &c.stack[len(c.stack)-1]
109. p, n := e.page, e.node

111. // If we have a node then search its inodes.
112. // 先搜索node
113. if n != nil {
114. //又是二分搜索
115. index := sort.Search(len(n.inodes), func(i int) bool {
116. return bytes.Compare(n.inodes[i].key, key) != -1
117. })
118. e.index = index
119. return
120. }

122. // If we have a page then search its leaf elements.
123. // 再搜索page
124. inodes := p.leafPageElements()
125. index := sort.Search(int(p.count), func(i int) bool {
126. return bytes.Compare(inodes[i].key(), key) != -1
127. })
128. e.index = index
129. }

到这儿我们就已经看完所有的seek()查找一个key的过程了，其内部也很简单，就是从根节点开始，通过不断递归遍历每层节点，采用二分法来定位到具体的叶子节点。到达叶子节点时，其叶子节点内部存储的数据也是有序的，因此继续按照二分查找来找到最终的下标。

值得需要注意点：

在遍历时，我们都知道，有可能遍历到的当前分支节点数据并没有在内存中，此时就需要从page中加载数据遍历。所以在遍历过程中，优先在node中找，如果node为空的时候才会采用page来查找。

3.2.4 First()、Last()实现分析

前面看了定位到具体某个key的一个过程，现在我们看一下，在定位到第一个元素时，我们知道它一定是位于最左侧的第一个叶子节点的第一个元素。同理，在定位到最后一个元素时，它一定是位于最右侧的第一个叶子节点的最后一个元素。下面是其内部的实现逻辑：

First()实现

1. // First moves the cursor to the first item in the bucket and returns its key and value.
2. // If the bucket is empty then a nil key and value are returned.
3. // The returned key and value are only valid for the life of the transaction.
4. func (c *Cursor) First() (key []byte, value []byte) {
5. _assert(c.bucket.tx.db != nil, "tx closed")
6. // 清空stack
7. c.stack = c.stack[:0]
8. p, n := c.bucket.pageNode(c.bucket.root)

10. // 一直找到第一个叶子节点，此处在天添加stack时，一直让index设置为0即可
11. ref := elemRef{page: p, node: n, index: 0}
12. c.stack = append(c.stack, ref)

15. c.first()

17. // If we land on an empty page then move to the next value.
18. // https://github.com/boltdb/bolt/issues/450
19. // 当前页时空的话，找下一个
20. if c.stack[len(c.stack)-1].count() == 0 {
21. c.next()
22. }

24. k, v, flags := c.keyValue()
25. // 是桶
26. if (flags & uint32(bucketLeafFlag)) != 0 {
27. return k, nil
28. }
29. return k, v

31. }

33. // first moves the cursor to the first leaf element under the last page in the stack.
34. // 找到最后一个非叶子节点的第一个叶子节点。index=0的节点
35. func (c *Cursor) first() {
36. for {
37. // Exit when we hit a leaf page.
38. var ref = &c.stack[len(c.stack)-1]
39. if ref.isLeaf() {
40. break
41. }

43. // Keep adding pages pointing to the first element to the stack.
44. var pgid pgid
45. if ref.node != nil {
46. pgid = ref.node.inodes[ref.index].pgid
47. } else {
48. pgid = ref.page.branchPageElement(uint16(ref.index)).pgid
49. }
50. p, n := c.bucket.pageNode(pgid)
51. c.stack = append(c.stack, elemRef{page: p, node: n, index: 0})
52. }
53. }

Last()实现

1. // Last moves the cursor to the last item in the bucket and returns its key and value.
2. // If the bucket is empty then a nil key and value are returned.
3. // The returned key and value are only valid for the life of the transaction.
4. func (c *Cursor) Last() (key []byte, value []byte) {
5. _assert(c.bucket.tx.db != nil, "tx closed")

7. c.stack = c.stack[:0]
8. p, n := c.bucket.pageNode(c.bucket.root)

10. ref := elemRef{page: p, node: n}
11. // 设置其index为当前页元素的最后一个
12. ref.index = ref.count() - 1
13. c.stack = append(c.stack, ref)

15. c.last()

17. k, v, flags := c.keyValue()
18. if (flags & uint32(bucketLeafFlag)) != 0 {
19. return k, nil
20. }
21. return k, v
22. }

24. // last moves the cursor to the last leaf element under the last page in the stack.
25. // 移动到栈中最后一个节点的最后一个叶子节点
26. func (c *Cursor) last() {
27. for {
28. // Exit when we hit a leaf page.
29. ref := &c.stack[len(c.stack)-1]
30. if ref.isLeaf() {
31. break
32. }

34. // Keep adding pages pointing to the last element in the stack.
35. var pgid pgid
36. if ref.node != nil {
37. pgid = ref.node.inodes[ref.index].pgid
38. } else {
39. pgid = ref.page.branchPageElement(uint16(ref.index)).pgid
40. }
41. p, n := c.bucket.pageNode(pgid)

43. var nextRef = elemRef{page: p, node: n}
44. nextRef.index = nextRef.count() - 1
45. c.stack = append(c.stack, nextRef)
46. }
47. }

3.2.5 Next()、Prev()实现分析

再此我们从当前位置查找前一个或者下一个时，需要注意一个问题，如果当前节点中元素已经完了，那么此时需要回退到遍历路径的上一个节点。然后再继续查找，下面进行代码分析。

Next()分析

1. // Next moves the cursor to the next item in the bucket and returns its key and value.
2. // If the cursor is at the end of the bucket then a nil key and value are returned.
3. // The returned key and value are only valid for the life of the transaction.
4. func (c *Cursor) Next() (key []byte, value []byte) {
5. _assert(c.bucket.tx.db != nil, "tx closed")
6. k, v, flags := c.next()
7. if (flags & uint32(bucketLeafFlag)) != 0 {
8. return k, nil
9. }
10. return k, v
11. }

13. // next moves to the next leaf element and returns the key and value.
14. // If the cursor is at the last leaf element then it stays there and returns nil.
15. func (c *Cursor) next() (key []byte, value []byte, flags uint32) {
16. for {
17. // Attempt to move over one element until we're successful.
18. // Move up the stack as we hit the end of each page in our stack.
19. var i int
20. for i = len(c.stack) - 1; i >= 0; i-- {
21. elem := &c.stack[i]
22. if elem.index < elem.count()-1 {
23. // 元素还有时，往后移动一个
24. elem.index++
25. break
26. }
27. }
28. // 最后的结果elem.index++

30. // If we've hit the root page then stop and return. This will leave the
31. // cursor on the last element of the last page.
32. // 所有页都遍历完了
33. if i == -1 {
34. return nil, nil, 0
35. }

37. // Otherwise start from where we left off in the stack and find the
38. // first element of the first leaf page.
39. // 剩余的节点里面找，跳过原先遍历过的节点
40. c.stack = c.stack[:i+1]
41. // 如果是叶子节点，first()啥都不做，直接退出。返回elem.index+1的数据
42. // 非叶子节点的话，需要移动到stack中最后一个路径的第一个元素
43. c.first()

45. // If this is an empty page then restart and move back up the stack.
46. // https://github.com/boltdb/bolt/issues/450
47. if c.stack[len(c.stack)-1].count() == 0 {
48. continue
49. }

51. return c.keyValue()
52. }
53. }

Prev()实现

1. // Prev moves the cursor to the previous item in the bucket and returns its key and value.
2. // If the cursor is at the beginning of the bucket then a nil key and value are returned.
3. // The returned key and value are only valid for the life of the transaction.
4. func (c *Cursor) Prev() (key []byte, value []byte) {
5. _assert(c.bucket.tx.db != nil, "tx closed")

7. // Attempt to move back one element until we're successful.
8. // Move up the stack as we hit the beginning of each page in our stack.
9. for i := len(c.stack) - 1; i >= 0; i-- {
10. elem := &c.stack[i]
11. if elem.index > 0 {
12. // 往前移动一格
13. elem.index--
14. break
15. }
16. c.stack = c.stack[:i]
17. }

19. // If we've hit the end then return nil.
20. if len(c.stack) == 0 {
21. return nil, nil
22. }

24. // Move down the stack to find the last element of the last leaf under this branch.
25. // 如果当前节点是叶子节点的话，则直接退出了，啥都不做。否则的话移动到新页的最后一个节点
26. c.last()
27. k, v, flags := c.keyValue()
28. if (flags & uint32(bucketLeafFlag)) != 0 {
29. return k, nil
30. }
31. return k, v
32. }

3.2.6 Delete()方法分析

Delete()方法中，移动当前位置的元素

1. // Delete removes the current key/value under the cursor from the bucket.
2. // Delete fails if current key/value is a bucket or if the transaction is not writable.
3. func (c *Cursor) Delete() error {
4. if c.bucket.tx.db == nil {
5. return ErrTxClosed
6. } else if !c.bucket.Writable() {
7. return ErrTxNotWritable
8. }

10. key, _, flags := c.keyValue()
11. // Return an error if current value is a bucket.
12. if (flags & bucketLeafFlag) != 0 {
13. return ErrIncompatibleValue
14. }
15. // 从node中移除，本质上将inode数组进行移动
16. c.node().del(key)

18. return nil
19. }

21. // del removes a key from the node.
22. func (n *node) del(key []byte) {
23. // Find index of key.
24. index := sort.Search(len(n.inodes), func(i int) bool { return bytes.Compare(n.inodes[i].key, key) != -1 })

26. // Exit if the key isn't found.
27. if index >= len(n.inodes) || !bytes.Equal(n.inodes[index].key, key) {
28. return
29. }

31. // Delete inode from the node.
32. n.inodes = append(n.inodes[:index], n.inodes[index+1:]...)

34. // Mark the node as needing rebalancing.
35. n.unbalanced = true
36. }