第四节 Bucket的相关操作

第四节 Bucket的相关操作

前面我们分析完了如何遍历、查找一个Bucket之后，下面我们来看看如何创建、获取、删除一个Bucket对象。

3.4.1 创建一个Bucket

1. CreateBucketIfNotExists()、CreateBucket()分析

根据指定的key来创建一个Bucket,如果指定key的Bucket已经存在，则会报错。如果指定的key之前有插入过元素，也会报错。否则的话，会在当前的Bucket中找到合适的位置，然后新建一个Bucket插入进去，最后返回给客户端。

2. // CreateBucketIfNotExists creates a new bucket if it doesn't already exist and returns a reference to it.
3. // Returns an error if the bucket name is blank, or if the bucket name is too long.
4. // The bucket instance is only valid for the lifetime of the transaction.
5. func (b *Bucket) CreateBucketIfNotExists(key []byte) (*Bucket, error) {
6. child, err := b.CreateBucket(key)
7. if err == ErrBucketExists {
8. return b.Bucket(key), nil
9. } else if err != nil {
10. return nil, err
11. }
12. return child, nil
13. }

15. // CreateBucket creates a new bucket at the given key and returns the new bucket.
16. // Returns an error if the key already exists, if the bucket name is blank, or if the bucket name is too long.
17. // The bucket instance is only valid for the lifetime of the transaction.
18. func (b *Bucket) CreateBucket(key []byte) (*Bucket, error) {
19. if b.tx.db == nil {
20. return nil, ErrTxClosed
21. } else if !b.tx.writable {
22. return nil, ErrTxNotWritable
23. } else if len(key) == 0 {
24. return nil, ErrBucketNameRequired
25. }

27. // Move cursor to correct position.
28. // 拿到游标
29. c := b.Cursor()
30. // 开始遍历、找到合适的位置
31. k, _, flags := c.seek(key)

33. // Return an error if there is an existing key.
34. if bytes.Equal(key, k) {
35. // 是桶,已经存在了
36. if (flags & bucketLeafFlag) != 0 {
37. return nil, ErrBucketExists
38. }
39. // 不是桶、但key已经存在了
40. return nil, ErrIncompatibleValue
41. }

43. // Create empty, inline bucket.
44. var bucket = Bucket{
45. bucket:      &bucket{},
46. rootNode:    &node{isLeaf: true},
47. FillPercent: DefaultFillPercent,
48. }
49. // 拿到bucket对应的value
50. var value = bucket.write()

52. // Insert into node.
53. key = cloneBytes(key)
54. // 插入到inode中
55. // c.node()方法会在内存中建立这棵树，调用n.read(page)
56. c.node().put(key, key, value, 0, bucketLeafFlag)

58. // Since subbuckets are not allowed on inline buckets, we need to
59. // dereference the inline page, if it exists. This will cause the bucket
60. // to be treated as a regular, non-inline bucket for the rest of the tx.
61. b.page = nil

63. //根据key获取一个桶
64. return b.Bucket(key), nil
65. }

67. // write allocates and writes a bucket to a byte slice.
68. // 内联桶的话，其value中bucketHeaderSize后面的内容为其page的数据
69. func (b *Bucket) write() []byte {
70. // Allocate the appropriate size.
71. var n = b.rootNode
72. var value = make([]byte, bucketHeaderSize+n.size())

74. // Write a bucket header.
75. var bucket = (*bucket)(unsafe.Pointer(&value[0]))
76. *bucket = *b.bucket

78. // Convert byte slice to a fake page and write the root node.
79. var p = (*page)(unsafe.Pointer(&value[bucketHeaderSize]))
80. // 将该桶中的元素压缩存储，放在value中
81. n.write(p)

83. return value
84. }

87. // node returns the node that the cursor is currently positioned on.
88. func (c *Cursor) node() *node {
89. _assert(len(c.stack) > 0, "accessing a node with a zero-length cursor stack")

91. // If the top of the stack is a leaf node then just return it.
92. if ref := &c.stack[len(c.stack)-1]; ref.node != nil && ref.isLeaf() {
93. return ref.node
94. }

96. // Start from root and traverse down the hierarchy.
97. var n = c.stack[0].node
98. if n == nil {
99. n = c.bucket.node(c.stack[0].page.id, nil)
100. }
101. // 非叶子节点
102. for _, ref := range c.stack[:len(c.stack)-1] {
103. _assert(!n.isLeaf, "expected branch node")
104. n = n.childAt(int(ref.index))
105. }
106. _assert(n.isLeaf, "expected leaf node")
107. return n
108. }

110. // put inserts a key/value.
111. // 如果put的是一个key、value的话，不需要指定pgid。
112. // 如果put的一个树枝节点，则需要指定pgid，不需要指定value
113. func (n *node) put(oldKey, newKey, value []byte, pgid pgid, flags uint32) {
114. if pgid >= n.bucket.tx.meta.pgid {
115. panic(fmt.Sprintf("pgid (%d) above high water mark (%d)", pgid, n.bucket.tx.meta.pgid))
116. } else if len(oldKey) <= 0 {
117. panic("put: zero-length old key")
118. } else if len(newKey) <= 0 {
119. panic("put: zero-length new key")
120. }

122. // Find insertion index.
123. index := sort.Search(len(n.inodes), func(i int) bool { return bytes.Compare(n.inodes[i].key, oldKey) != -1 })

125. // Add capacity and shift nodes if we don't have an exact match and need to insert.
126. exact := (len(n.inodes) > 0 && index < len(n.inodes) && bytes.Equal(n.inodes[index].key, oldKey))
127. if !exact {
128. n.inodes = append(n.inodes, inode{})
129. copy(n.inodes[index+1:], n.inodes[index:])
130. }

132. inode := &n.inodes[index]
133. inode.flags = flags
134. inode.key = newKey
135. inode.value = value
136. inode.pgid = pgid
137. _assert(len(inode.key) > 0, "put: zero-length inode key")
138. }

3.4.2 获取一个Bucket

根据指定的key来获取一个Bucket。如果找不到则返回nil。

1. // Bucket retrieves a nested bucket by name.
2. // Returns nil if the bucket does not exist.
3. // The bucket instance is only valid for the lifetime of the transaction.
4. func (b *Bucket) Bucket(name []byte) *Bucket {
5. if b.buckets != nil {
6. if child := b.buckets[string(name)]; child != nil {
7. return child
8. }
9. }

11. // Move cursor to key.
12. // 根据游标找key
13. c := b.Cursor()
14. k, v, flags := c.seek(name)

16. // Return nil if the key doesn't exist or it is not a bucket.
17. if !bytes.Equal(name, k) || (flags&bucketLeafFlag) == 0 {
18. return nil
19. }

21. // Otherwise create a bucket and cache it.
22. // 根据找到的value来打开桶。
23. var child = b.openBucket(v)
24. // 加速缓存的作用
25. if b.buckets != nil {
26. b.buckets[string(name)] = child
27. }

29. return child
30. }

32. // Helper method that re-interprets a sub-bucket value
33. // from a parent into a Bucket
34. func (b *Bucket) openBucket(value []byte) *Bucket {
35. var child = newBucket(b.tx)

37. // If unaligned load/stores are broken on this arch and value is
38. // unaligned simply clone to an aligned byte array.
39. unaligned := brokenUnaligned && uintptr(unsafe.Pointer(&value[0]))&3 != 0

41. if unaligned {
42. value = cloneBytes(value)
43. }

45. // If this is a writable transaction then we need to copy the bucket entry.
46. // Read-only transactions can point directly at the mmap entry.
47. if b.tx.writable && !unaligned {
48. child.bucket = &bucket{}
49. *child.bucket = *(*bucket)(unsafe.Pointer(&value[0]))
50. } else {
51. child.bucket = (*bucket)(unsafe.Pointer(&value[0]))
52. }

54. // Save a reference to the inline page if the bucket is inline.
55. // 内联桶
56. if child.root == 0 {
57. child.page = (*page)(unsafe.Pointer(&value[bucketHeaderSize]))
58. }

60. return &child
61. }

3.4.3 删除一个Bucket

DeleteBucket()方法用来删除一个指定key的Bucket。其内部实现逻辑是先递归的删除其子桶。然后再释放该Bucket的page，并最终从叶子节点中移除

1. // DeleteBucket deletes a bucket at the given key.
2. // Returns an error if the bucket does not exists, or if the key represents a non-bucket value.
3. func (b *Bucket) DeleteBucket(key []byte) error {
4. if b.tx.db == nil {
5. return ErrTxClosed
6. } else if !b.Writable() {
7. return ErrTxNotWritable
8. }

10. // Move cursor to correct position.
11. c := b.Cursor()
12. k, _, flags := c.seek(key)

14. // Return an error if bucket doesn't exist or is not a bucket.
15. if !bytes.Equal(key, k) {
16. return ErrBucketNotFound
17. } else if (flags & bucketLeafFlag) == 0 {
18. return ErrIncompatibleValue
19. }

21. // Recursively delete all child buckets.
22. child := b.Bucket(key)
23. // 将该桶下面的所有桶都删除
24. err := child.ForEach(func(k, v []byte) error {
25. if v == nil {
26. if err := child.DeleteBucket(k); err != nil {
27. return fmt.Errorf("delete bucket: %s", err)
28. }
29. }
30. return nil
31. })
32. if err != nil {
33. return err
34. }

36. // Remove cached copy.
37. delete(b.buckets, string(key))

39. // Release all bucket pages to freelist.
40. child.nodes = nil
41. child.rootNode = nil
42. child.free()

44. // Delete the node if we have a matching key.
45. c.node().del(key)

47. return nil
48. }

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

55. // Exit if the key isn't found.
56. if index >= len(n.inodes) || !bytes.Equal(n.inodes[index].key, key) {
57. return
58. }

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

63. // Mark the node as needing rebalancing.
64. n.unbalanced = true
65. }

67. // free recursively frees all pages in the bucket.
68. func (b *Bucket) free() {
69. if b.root == 0 {
70. return
71. }

73. var tx = b.tx
74. b.forEachPageNode(func(p *page, n *node, _ int) {
75. if p != nil {
76. tx.db.freelist.free(tx.meta.txid, p)
77. } else {
78. n.free()
79. }
80. })
81. b.root = 0
82. }