第四节 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. }