Refactor leaf.write() / leaf.split().

1 files changed, 61 insertions, 74 deletions

diff --git a/leaf.go b/leaf.go
index 23b35eb..dcd343f 100644
--- a/leaf.go
+++ b/leaf.go
@@ -9,16 +9,10 @@ import (
 // leaf represents a temporary in-memory leaf page.
 // It is deserialized from an memory-mapped page and is not restricted by page size.
 type leaf struct {
+	parent *branch
 	items leafItems
 }
 
-type leafItems []leafItem
-
-type leafItem struct {
-	key   []byte
-	value []byte
-}
-
 // put inserts or replaces a key on a leaf page.
 func (l *leaf) put(key []byte, value []byte) {
 	// Find insertion index.
@@ -35,11 +29,20 @@ func (l *leaf) put(key []byte, value []byte) {
 	l.items[index].value = value
 }
 
+// size returns the size of the leaf after serialization.
+func (l *leaf) size() int {
+	var size int = pageHeaderSize
+	for _, item := range l.items {
+		size += lnodeSize + len(item.key) + len(item.value)
+	}
+	return size
+}
+
 // read initializes the item data from an on-disk page.
-func (l *leaf) read(page *page) {
-	ncount := int(page.count)
+func (l *leaf) read(p *page) {
+	ncount := int(p.count)
 	l.items = make(leafItems, ncount)
-	lnodes := (*[maxNodesPerPage]lnode)(unsafe.Pointer(&page.ptr))
+	lnodes := (*[maxNodesPerPage]lnode)(unsafe.Pointer(&p.ptr))
 	for i := 0; i < ncount; i++ {
 		lnode := &lnodes[i]
 		item := &l.items[i]
@@ -49,86 +52,70 @@ func (l *leaf) read(page *page) {
 }
 
 // write writes the items onto one or more leaf pages.
-func (l *leaf) write(pageSize int, allocate func(size int) (*page, error)) ([]*page, error) {
-	var pages []*page
-
-	for _, items := range l.split(pageSize) {
-		// Determine the total page size.
-		var size int = pageHeaderSize
-		for _, item := range l.items {
-			size += lnodeSize + len(item.key) + len(item.value)
-		}
-
-		// Allocate pages.
-		page, err := allocate(size)
-		if err != nil {
-			return nil, err
-		}
-		page.flags |= p_leaf
-		page.count = uint16(len(items))
-
-		// Loop over each item and write it to the page.
-		lnodes := (*[maxNodesPerPage]lnode)(unsafe.Pointer(&page.ptr))
-		b := (*[maxAllocSize]byte)(unsafe.Pointer(&page.ptr))[lnodeSize*len(items):]
-		for index, item := range items {
-			// Write item.
-			lnode := &lnodes[index]
-			lnode.pos = uint32(uintptr(unsafe.Pointer(&b[0])) - uintptr(unsafe.Pointer(lnode)))
-			lnode.ksize = uint32(len(item.key))
-			lnode.vsize = uint32(len(item.value))
-
-			// Write data to the end of the page.
-			copy(b[0:], item.key)
-			b = b[len(item.key):]
-			copy(b[0:], item.value)
-			b = b[len(item.value):]
-		}
-
-		pages = append(pages, page)
+func (l *leaf) write(p *page) {
+	// Initialize page.
+	p.flags |= p_leaf
+	p.count = uint16(len(l.items))
+
+	// Loop over each item and write it to the page.
+	lnodes := (*[maxNodesPerPage]lnode)(unsafe.Pointer(&p.ptr))
+	b := (*[maxAllocSize]byte)(unsafe.Pointer(&p.ptr))[lnodeSize*len(l.items):]
+	for index, item := range l.items {
+		// Write node.
+		lnode := &lnodes[index]
+		lnode.pos = uint32(uintptr(unsafe.Pointer(&b[0])) - uintptr(unsafe.Pointer(lnode)))
+		lnode.ksize = uint32(len(item.key))
+		lnode.vsize = uint32(len(item.value))
+
+		// Write data to the end of the page.
+		copy(b[0:], item.key)
+		b = b[len(item.key):]
+		copy(b[0:], item.value)
+		b = b[len(item.value):]
 	}
-
-	return pages, nil
 }
 
 // split divides up the noes in the page into appropriately sized groups.
-func (l *leaf) split(pageSize int) []leafItems {
-	// If we don't have enough items for multiple pages then just return the items.
-	if len(l.items) <= (minKeysPerPage * 2) {
-		return []leafItems{l.items}
+func (l *leaf) split(pageSize int) []*leaf {
+	// Ignore the split if the page doesn't have at least enough nodes for
+	// multiple pages or if the data can fit on a single page.
+	if len(l.items) <= (minKeysPerPage * 2) || l.size() < pageSize {
+		return []*leaf{l}
 	}
 
-	// If we're not larger than one page then just return the items.
-	var totalSize int = pageHeaderSize
-	for _, item := range l.items {
-		totalSize += lnodeSize + len(item.key) + len(item.value)
-	}
-	if totalSize < pageSize {
-		return []leafItems{l.items}
-	}
-
-	// Otherwise group into smaller pages and target a given fill size.
-	var size int
-	var group leafItems
-	var groups []leafItems
-
 	// Set fill threshold to 25%.
 	threshold := pageSize >> 4
 
+	// Otherwise group into smaller pages and target a given fill size.
+	size := 0
+	current := &leaf{}
+	leafs := make([]*leaf, 0)
+
 	for index, item := range l.items {
 		nodeSize := lnodeSize + len(item.key) + len(item.value)
 
-		if group == nil || (len(group) >= minKeysPerPage && index < len(l.items)-minKeysPerPage && size+nodeSize > threshold) {
+		if (len(current.items) >= minKeysPerPage && index < len(l.items)-minKeysPerPage && size+nodeSize > threshold) {
 			size = pageHeaderSize
-			if group != nil {
-				groups = append(groups, group)
-			}
-			group = make(leafItems, 0)
+			leafs = append(leafs, current)
+			current = &leaf{}
 		}
 
 		size += nodeSize
-		group = append(group, item)
+		current.items = append(current.items, item)
 	}
-	groups = append(groups, group)
+	leafs = append(leafs, current)
 
-	return groups
+	return leafs
 }
+
+type leafItems []leafItem
+
+type leafItem struct {
+	key   []byte
+	value []byte
+}
+
+func (s leafItems) Len() int           { return len(s) }
+func (s leafItems) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
+func (s leafItems) Less(i, j int) bool { return bytes.Compare(s[i].key, s[j].key) == -1 }
+