1package bolt 2 3import ( 4 "bytes" 5 "fmt" 6 "unsafe" 7) 8 9const ( 10 // MaxKeySize is the maximum length of a key, in bytes. 11 MaxKeySize = 32768 12 13 // MaxValueSize is the maximum length of a value, in bytes. 14 MaxValueSize = (1 << 31) - 2 15) 16 17const ( 18 maxUint = ^uint(0) 19 minUint = 0 20 maxInt = int(^uint(0) >> 1) 21 minInt = -maxInt - 1 22) 23 24const bucketHeaderSize = int(unsafe.Sizeof(bucket{})) 25 26const ( 27 minFillPercent = 0.1 28 maxFillPercent = 1.0 29) 30 31// DefaultFillPercent is the percentage that split pages are filled. 32// This value can be changed by setting Bucket.FillPercent. 33const DefaultFillPercent = 0.5 34 35// Bucket represents a collection of key/value pairs inside the database. 36type Bucket struct { 37 *bucket 38 tx *Tx // the associated transaction 39 buckets map[string]*Bucket // subbucket cache 40 page *page // inline page reference 41 rootNode *node // materialized node for the root page. 42 nodes map[pgid]*node // node cache 43 44 // Sets the threshold for filling nodes when they split. By default, 45 // the bucket will fill to 50% but it can be useful to increase this 46 // amount if you know that your write workloads are mostly append-only. 47 // 48 // This is non-persisted across transactions so it must be set in every Tx. 49 FillPercent float64 50} 51 52// bucket represents the on-file representation of a bucket. 53// This is stored as the "value" of a bucket key. If the bucket is small enough, 54// then its root page can be stored inline in the "value", after the bucket 55// header. In the case of inline buckets, the "root" will be 0. 56type bucket struct { 57 root pgid // page id of the bucket's root-level page 58 sequence uint64 // monotonically incrementing, used by NextSequence() 59} 60 61// newBucket returns a new bucket associated with a transaction. 62func newBucket(tx *Tx) Bucket { 63 var b = Bucket{tx: tx, FillPercent: DefaultFillPercent} 64 if tx.writable { 65 b.buckets = make(map[string]*Bucket) 66 b.nodes = make(map[pgid]*node) 67 } 68 return b 69} 70 71// Tx returns the tx of the bucket. 72func (b *Bucket) Tx() *Tx { 73 return b.tx 74} 75 76// Root returns the root of the bucket. 77func (b *Bucket) Root() pgid { 78 return b.root 79} 80 81// Writable returns whether the bucket is writable. 82func (b *Bucket) Writable() bool { 83 return b.tx.writable 84} 85 86// Cursor creates a cursor associated with the bucket. 87// The cursor is only valid as long as the transaction is open. 88// Do not use a cursor after the transaction is closed. 89func (b *Bucket) Cursor() *Cursor { 90 // Update transaction statistics. 91 b.tx.stats.CursorCount++ 92 93 // Allocate and return a cursor. 94 return &Cursor{ 95 bucket: b, 96 stack: make([]elemRef, 0), 97 } 98} 99 100// Bucket retrieves a nested bucket by name. 101// Returns nil if the bucket does not exist. 102// The bucket instance is only valid for the lifetime of the transaction. 103func (b *Bucket) Bucket(name []byte) *Bucket { 104 if b.buckets != nil { 105 if child := b.buckets[string(name)]; child != nil { 106 return child 107 } 108 } 109 110 // Move cursor to key. 111 c := b.Cursor() 112 k, v, flags := c.seek(name) 113 114 // Return nil if the key doesn't exist or it is not a bucket. 115 if !bytes.Equal(name, k) || (flags&bucketLeafFlag) == 0 { 116 return nil 117 } 118 119 // Otherwise create a bucket and cache it. 120 var child = b.openBucket(v) 121 if b.buckets != nil { 122 b.buckets[string(name)] = child 123 } 124 125 return child 126} 127 128// Helper method that re-interprets a sub-bucket value 129// from a parent into a Bucket 130func (b *Bucket) openBucket(value []byte) *Bucket { 131 var child = newBucket(b.tx) 132 133 // If this is a writable transaction then we need to copy the bucket entry. 134 // Read-only transactions can point directly at the mmap entry. 135 if b.tx.writable { 136 child.bucket = &bucket{} 137 *child.bucket = *(*bucket)(unsafe.Pointer(&value[0])) 138 } else { 139 child.bucket = (*bucket)(unsafe.Pointer(&value[0])) 140 } 141 142 // Save a reference to the inline page if the bucket is inline. 143 if child.root == 0 { 144 child.page = (*page)(unsafe.Pointer(&value[bucketHeaderSize])) 145 } 146 147 return &child 148} 149 150// CreateBucket creates a new bucket at the given key and returns the new bucket. 151// Returns an error if the key already exists, if the bucket name is blank, or if the bucket name is too long. 152// The bucket instance is only valid for the lifetime of the transaction. 153func (b *Bucket) CreateBucket(key []byte) (*Bucket, error) { 154 if b.tx.db == nil { 155 return nil, ErrTxClosed 156 } else if !b.tx.writable { 157 return nil, ErrTxNotWritable 158 } else if len(key) == 0 { 159 return nil, ErrBucketNameRequired 160 } 161 162 // Move cursor to correct position. 163 c := b.Cursor() 164 k, _, flags := c.seek(key) 165 166 // Return an error if there is an existing key. 167 if bytes.Equal(key, k) { 168 if (flags & bucketLeafFlag) != 0 { 169 return nil, ErrBucketExists 170 } else { 171 return nil, ErrIncompatibleValue 172 } 173 } 174 175 // Create empty, inline bucket. 176 var bucket = Bucket{ 177 bucket: &bucket{}, 178 rootNode: &node{isLeaf: true}, 179 FillPercent: DefaultFillPercent, 180 } 181 var value = bucket.write() 182 183 // Insert into node. 184 key = cloneBytes(key) 185 c.node().put(key, key, value, 0, bucketLeafFlag) 186 187 // Since subbuckets are not allowed on inline buckets, we need to 188 // dereference the inline page, if it exists. This will cause the bucket 189 // to be treated as a regular, non-inline bucket for the rest of the tx. 190 b.page = nil 191 192 return b.Bucket(key), nil 193} 194 195// CreateBucketIfNotExists creates a new bucket if it doesn't already exist and returns a reference to it. 196// Returns an error if the bucket name is blank, or if the bucket name is too long. 197// The bucket instance is only valid for the lifetime of the transaction. 198func (b *Bucket) CreateBucketIfNotExists(key []byte) (*Bucket, error) { 199 child, err := b.CreateBucket(key) 200 if err == ErrBucketExists { 201 return b.Bucket(key), nil 202 } else if err != nil { 203 return nil, err 204 } 205 return child, nil 206} 207 208// DeleteBucket deletes a bucket at the given key. 209// Returns an error if the bucket does not exists, or if the key represents a non-bucket value. 210func (b *Bucket) DeleteBucket(key []byte) error { 211 if b.tx.db == nil { 212 return ErrTxClosed 213 } else if !b.Writable() { 214 return ErrTxNotWritable 215 } 216 217 // Move cursor to correct position. 218 c := b.Cursor() 219 k, _, flags := c.seek(key) 220 221 // Return an error if bucket doesn't exist or is not a bucket. 222 if !bytes.Equal(key, k) { 223 return ErrBucketNotFound 224 } else if (flags & bucketLeafFlag) == 0 { 225 return ErrIncompatibleValue 226 } 227 228 // Recursively delete all child buckets. 229 child := b.Bucket(key) 230 err := child.ForEach(func(k, v []byte) error { 231 if v == nil { 232 if err := child.DeleteBucket(k); err != nil { 233 return fmt.Errorf("delete bucket: %s", err) 234 } 235 } 236 return nil 237 }) 238 if err != nil { 239 return err 240 } 241 242 // Remove cached copy. 243 delete(b.buckets, string(key)) 244 245 // Release all bucket pages to freelist. 246 child.nodes = nil 247 child.rootNode = nil 248 child.free() 249 250 // Delete the node if we have a matching key. 251 c.node().del(key) 252 253 return nil 254} 255 256// Get retrieves the value for a key in the bucket. 257// Returns a nil value if the key does not exist or if the key is a nested bucket. 258// The returned value is only valid for the life of the transaction. 259func (b *Bucket) Get(key []byte) []byte { 260 k, v, flags := b.Cursor().seek(key) 261 262 // Return nil if this is a bucket. 263 if (flags & bucketLeafFlag) != 0 { 264 return nil 265 } 266 267 // If our target node isn't the same key as what's passed in then return nil. 268 if !bytes.Equal(key, k) { 269 return nil 270 } 271 return v 272} 273 274// Put sets the value for a key in the bucket. 275// If the key exist then its previous value will be overwritten. 276// Supplied value must remain valid for the life of the transaction. 277// Returns an error if the bucket was created from a read-only transaction, if the key is blank, if the key is too large, or if the value is too large. 278func (b *Bucket) Put(key []byte, value []byte) error { 279 if b.tx.db == nil { 280 return ErrTxClosed 281 } else if !b.Writable() { 282 return ErrTxNotWritable 283 } else if len(key) == 0 { 284 return ErrKeyRequired 285 } else if len(key) > MaxKeySize { 286 return ErrKeyTooLarge 287 } else if int64(len(value)) > MaxValueSize { 288 return ErrValueTooLarge 289 } 290 291 // Move cursor to correct position. 292 c := b.Cursor() 293 k, _, flags := c.seek(key) 294 295 // Return an error if there is an existing key with a bucket value. 296 if bytes.Equal(key, k) && (flags&bucketLeafFlag) != 0 { 297 return ErrIncompatibleValue 298 } 299 300 // Insert into node. 301 key = cloneBytes(key) 302 c.node().put(key, key, value, 0, 0) 303 304 return nil 305} 306 307// Delete removes a key from the bucket. 308// If the key does not exist then nothing is done and a nil error is returned. 309// Returns an error if the bucket was created from a read-only transaction. 310func (b *Bucket) Delete(key []byte) error { 311 if b.tx.db == nil { 312 return ErrTxClosed 313 } else if !b.Writable() { 314 return ErrTxNotWritable 315 } 316 317 // Move cursor to correct position. 318 c := b.Cursor() 319 _, _, flags := c.seek(key) 320 321 // Return an error if there is already existing bucket value. 322 if (flags & bucketLeafFlag) != 0 { 323 return ErrIncompatibleValue 324 } 325 326 // Delete the node if we have a matching key. 327 c.node().del(key) 328 329 return nil 330} 331 332// NextSequence returns an autoincrementing integer for the bucket. 333func (b *Bucket) NextSequence() (uint64, error) { 334 if b.tx.db == nil { 335 return 0, ErrTxClosed 336 } else if !b.Writable() { 337 return 0, ErrTxNotWritable 338 } 339 340 // Materialize the root node if it hasn't been already so that the 341 // bucket will be saved during commit. 342 if b.rootNode == nil { 343 _ = b.node(b.root, nil) 344 } 345 346 // Increment and return the sequence. 347 b.bucket.sequence++ 348 return b.bucket.sequence, nil 349} 350 351// ForEach executes a function for each key/value pair in a bucket. 352// If the provided function returns an error then the iteration is stopped and 353// the error is returned to the caller. The provided function must not modify 354// the bucket; this will result in undefined behavior. 355func (b *Bucket) ForEach(fn func(k, v []byte) error) error { 356 if b.tx.db == nil { 357 return ErrTxClosed 358 } 359 c := b.Cursor() 360 for k, v := c.First(); k != nil; k, v = c.Next() { 361 if err := fn(k, v); err != nil { 362 return err 363 } 364 } 365 return nil 366} 367 368// Stat returns stats on a bucket. 369func (b *Bucket) Stats() BucketStats { 370 var s, subStats BucketStats 371 pageSize := b.tx.db.pageSize 372 s.BucketN += 1 373 if b.root == 0 { 374 s.InlineBucketN += 1 375 } 376 b.forEachPage(func(p *page, depth int) { 377 if (p.flags & leafPageFlag) != 0 { 378 s.KeyN += int(p.count) 379 380 // used totals the used bytes for the page 381 used := pageHeaderSize 382 383 if p.count != 0 { 384 // If page has any elements, add all element headers. 385 used += leafPageElementSize * int(p.count-1) 386 387 // Add all element key, value sizes. 388 // The computation takes advantage of the fact that the position 389 // of the last element's key/value equals to the total of the sizes 390 // of all previous elements' keys and values. 391 // It also includes the last element's header. 392 lastElement := p.leafPageElement(p.count - 1) 393 used += int(lastElement.pos + lastElement.ksize + lastElement.vsize) 394 } 395 396 if b.root == 0 { 397 // For inlined bucket just update the inline stats 398 s.InlineBucketInuse += used 399 } else { 400 // For non-inlined bucket update all the leaf stats 401 s.LeafPageN++ 402 s.LeafInuse += used 403 s.LeafOverflowN += int(p.overflow) 404 405 // Collect stats from sub-buckets. 406 // Do that by iterating over all element headers 407 // looking for the ones with the bucketLeafFlag. 408 for i := uint16(0); i < p.count; i++ { 409 e := p.leafPageElement(i) 410 if (e.flags & bucketLeafFlag) != 0 { 411 // For any bucket element, open the element value 412 // and recursively call Stats on the contained bucket. 413 subStats.Add(b.openBucket(e.value()).Stats()) 414 } 415 } 416 } 417 } else if (p.flags & branchPageFlag) != 0 { 418 s.BranchPageN++ 419 lastElement := p.branchPageElement(p.count - 1) 420 421 // used totals the used bytes for the page 422 // Add header and all element headers. 423 used := pageHeaderSize + (branchPageElementSize * int(p.count-1)) 424 425 // Add size of all keys and values. 426 // Again, use the fact that last element's position equals to 427 // the total of key, value sizes of all previous elements. 428 used += int(lastElement.pos + lastElement.ksize) 429 s.BranchInuse += used 430 s.BranchOverflowN += int(p.overflow) 431 } 432 433 // Keep track of maximum page depth. 434 if depth+1 > s.Depth { 435 s.Depth = (depth + 1) 436 } 437 }) 438 439 // Alloc stats can be computed from page counts and pageSize. 440 s.BranchAlloc = (s.BranchPageN + s.BranchOverflowN) * pageSize 441 s.LeafAlloc = (s.LeafPageN + s.LeafOverflowN) * pageSize 442 443 // Add the max depth of sub-buckets to get total nested depth. 444 s.Depth += subStats.Depth 445 // Add the stats for all sub-buckets 446 s.Add(subStats) 447 return s 448} 449 450// forEachPage iterates over every page in a bucket, including inline pages. 451func (b *Bucket) forEachPage(fn func(*page, int)) { 452 // If we have an inline page then just use that. 453 if b.page != nil { 454 fn(b.page, 0) 455 return 456 } 457 458 // Otherwise traverse the page hierarchy. 459 b.tx.forEachPage(b.root, 0, fn) 460} 461 462// forEachPageNode iterates over every page (or node) in a bucket. 463// This also includes inline pages. 464func (b *Bucket) forEachPageNode(fn func(*page, *node, int)) { 465 // If we have an inline page or root node then just use that. 466 if b.page != nil { 467 fn(b.page, nil, 0) 468 return 469 } 470 b._forEachPageNode(b.root, 0, fn) 471} 472 473func (b *Bucket) _forEachPageNode(pgid pgid, depth int, fn func(*page, *node, int)) { 474 var p, n = b.pageNode(pgid) 475 476 // Execute function. 477 fn(p, n, depth) 478 479 // Recursively loop over children. 480 if p != nil { 481 if (p.flags & branchPageFlag) != 0 { 482 for i := 0; i < int(p.count); i++ { 483 elem := p.branchPageElement(uint16(i)) 484 b._forEachPageNode(elem.pgid, depth+1, fn) 485 } 486 } 487 } else { 488 if !n.isLeaf { 489 for _, inode := range n.inodes { 490 b._forEachPageNode(inode.pgid, depth+1, fn) 491 } 492 } 493 } 494} 495 496// spill writes all the nodes for this bucket to dirty pages. 497func (b *Bucket) spill() error { 498 // Spill all child buckets first. 499 for name, child := range b.buckets { 500 // If the child bucket is small enough and it has no child buckets then 501 // write it inline into the parent bucket's page. Otherwise spill it 502 // like a normal bucket and make the parent value a pointer to the page. 503 var value []byte 504 if child.inlineable() { 505 child.free() 506 value = child.write() 507 } else { 508 if err := child.spill(); err != nil { 509 return err 510 } 511 512 // Update the child bucket header in this bucket. 513 value = make([]byte, unsafe.Sizeof(bucket{})) 514 var bucket = (*bucket)(unsafe.Pointer(&value[0])) 515 *bucket = *child.bucket 516 } 517 518 // Skip writing the bucket if there are no materialized nodes. 519 if child.rootNode == nil { 520 continue 521 } 522 523 // Update parent node. 524 var c = b.Cursor() 525 k, _, flags := c.seek([]byte(name)) 526 if !bytes.Equal([]byte(name), k) { 527 panic(fmt.Sprintf("misplaced bucket header: %x -> %x", []byte(name), k)) 528 } 529 if flags&bucketLeafFlag == 0 { 530 panic(fmt.Sprintf("unexpected bucket header flag: %x", flags)) 531 } 532 c.node().put([]byte(name), []byte(name), value, 0, bucketLeafFlag) 533 } 534 535 // Ignore if there's not a materialized root node. 536 if b.rootNode == nil { 537 return nil 538 } 539 540 // Spill nodes. 541 if err := b.rootNode.spill(); err != nil { 542 return err 543 } 544 b.rootNode = b.rootNode.root() 545 546 // Update the root node for this bucket. 547 if b.rootNode.pgid >= b.tx.meta.pgid { 548 panic(fmt.Sprintf("pgid (%d) above high water mark (%d)", b.rootNode.pgid, b.tx.meta.pgid)) 549 } 550 b.root = b.rootNode.pgid 551 552 return nil 553} 554 555// inlineable returns true if a bucket is small enough to be written inline 556// and if it contains no subbuckets. Otherwise returns false. 557func (b *Bucket) inlineable() bool { 558 var n = b.rootNode 559 560 // Bucket must only contain a single leaf node. 561 if n == nil || !n.isLeaf { 562 return false 563 } 564 565 // Bucket is not inlineable if it contains subbuckets or if it goes beyond 566 // our threshold for inline bucket size. 567 var size = pageHeaderSize 568 for _, inode := range n.inodes { 569 size += leafPageElementSize + len(inode.key) + len(inode.value) 570 571 if inode.flags&bucketLeafFlag != 0 { 572 return false 573 } else if size > b.maxInlineBucketSize() { 574 return false 575 } 576 } 577 578 return true 579} 580 581// Returns the maximum total size of a bucket to make it a candidate for inlining. 582func (b *Bucket) maxInlineBucketSize() int { 583 return b.tx.db.pageSize / 4 584} 585 586// write allocates and writes a bucket to a byte slice. 587func (b *Bucket) write() []byte { 588 // Allocate the appropriate size. 589 var n = b.rootNode 590 var value = make([]byte, bucketHeaderSize+n.size()) 591 592 // Write a bucket header. 593 var bucket = (*bucket)(unsafe.Pointer(&value[0])) 594 *bucket = *b.bucket 595 596 // Convert byte slice to a fake page and write the root node. 597 var p = (*page)(unsafe.Pointer(&value[bucketHeaderSize])) 598 n.write(p) 599 600 return value 601} 602 603// rebalance attempts to balance all nodes. 604func (b *Bucket) rebalance() { 605 for _, n := range b.nodes { 606 n.rebalance() 607 } 608 for _, child := range b.buckets { 609 child.rebalance() 610 } 611} 612 613// node creates a node from a page and associates it with a given parent. 614func (b *Bucket) node(pgid pgid, parent *node) *node { 615 _assert(b.nodes != nil, "nodes map expected") 616 617 // Retrieve node if it's already been created. 618 if n := b.nodes[pgid]; n != nil { 619 return n 620 } 621 622 // Otherwise create a node and cache it. 623 n := &node{bucket: b, parent: parent} 624 if parent == nil { 625 b.rootNode = n 626 } else { 627 parent.children = append(parent.children, n) 628 } 629 630 // Use the inline page if this is an inline bucket. 631 var p = b.page 632 if p == nil { 633 p = b.tx.page(pgid) 634 } 635 636 // Read the page into the node and cache it. 637 n.read(p) 638 b.nodes[pgid] = n 639 640 // Update statistics. 641 b.tx.stats.NodeCount++ 642 643 return n 644} 645 646// free recursively frees all pages in the bucket. 647func (b *Bucket) free() { 648 if b.root == 0 { 649 return 650 } 651 652 var tx = b.tx 653 b.forEachPageNode(func(p *page, n *node, _ int) { 654 if p != nil { 655 tx.db.freelist.free(tx.meta.txid, p) 656 } else { 657 n.free() 658 } 659 }) 660 b.root = 0 661} 662 663// dereference removes all references to the old mmap. 664func (b *Bucket) dereference() { 665 if b.rootNode != nil { 666 b.rootNode.root().dereference() 667 } 668 669 for _, child := range b.buckets { 670 child.dereference() 671 } 672} 673 674// pageNode returns the in-memory node, if it exists. 675// Otherwise returns the underlying page. 676func (b *Bucket) pageNode(id pgid) (*page, *node) { 677 // Inline buckets have a fake page embedded in their value so treat them 678 // differently. We'll return the rootNode (if available) or the fake page. 679 if b.root == 0 { 680 if id != 0 { 681 panic(fmt.Sprintf("inline bucket non-zero page access(2): %d != 0", id)) 682 } 683 if b.rootNode != nil { 684 return nil, b.rootNode 685 } 686 return b.page, nil 687 } 688 689 // Check the node cache for non-inline buckets. 690 if b.nodes != nil { 691 if n := b.nodes[id]; n != nil { 692 return nil, n 693 } 694 } 695 696 // Finally lookup the page from the transaction if no node is materialized. 697 return b.tx.page(id), nil 698} 699 700// BucketStats records statistics about resources used by a bucket. 701type BucketStats struct { 702 // Page count statistics. 703 BranchPageN int // number of logical branch pages 704 BranchOverflowN int // number of physical branch overflow pages 705 LeafPageN int // number of logical leaf pages 706 LeafOverflowN int // number of physical leaf overflow pages 707 708 // Tree statistics. 709 KeyN int // number of keys/value pairs 710 Depth int // number of levels in B+tree 711 712 // Page size utilization. 713 BranchAlloc int // bytes allocated for physical branch pages 714 BranchInuse int // bytes actually used for branch data 715 LeafAlloc int // bytes allocated for physical leaf pages 716 LeafInuse int // bytes actually used for leaf data 717 718 // Bucket statistics 719 BucketN int // total number of buckets including the top bucket 720 InlineBucketN int // total number on inlined buckets 721 InlineBucketInuse int // bytes used for inlined buckets (also accounted for in LeafInuse) 722} 723 724func (s *BucketStats) Add(other BucketStats) { 725 s.BranchPageN += other.BranchPageN 726 s.BranchOverflowN += other.BranchOverflowN 727 s.LeafPageN += other.LeafPageN 728 s.LeafOverflowN += other.LeafOverflowN 729 s.KeyN += other.KeyN 730 if s.Depth < other.Depth { 731 s.Depth = other.Depth 732 } 733 s.BranchAlloc += other.BranchAlloc 734 s.BranchInuse += other.BranchInuse 735 s.LeafAlloc += other.LeafAlloc 736 s.LeafInuse += other.LeafInuse 737 738 s.BucketN += other.BucketN 739 s.InlineBucketN += other.InlineBucketN 740 s.InlineBucketInuse += other.InlineBucketInuse 741} 742 743// cloneBytes returns a copy of a given slice. 744func cloneBytes(v []byte) []byte { 745 var clone = make([]byte, len(v)) 746 copy(clone, v) 747 return clone 748} 749