Implement optimized fast leaf write.

This commit adds an optimized implementation for `putLeafCell()` if
the insert/update will not cause the page to overflow.

rbf/cursor.go

@@ -349,14 +349,32 @@ func fromPgno(val uint32) []byte {
 func toPgno(val []byte) uint32 {
 	return binary.LittleEndian.Uint32(val)
 }
+
 func (c *Cursor) putLeafCell(in leafCell) (err error) {
 
 	leafPage := c.leafPage // the last read leaf page
-	cells := readLeafCells(leafPage, c.leafCells[:])
 	elem := &c.stack.elems[c.stack.index]
-	cell := in
-	if elem.index >= len(cells) || c.Key() != cell.Key {
+	cellN := readCellN(leafPage)
 
+	// Determine if the insert/update will overflow the page.
+	// If it doesn't then we can do an optimized write where we don't deserialize.
+	isInsert := elem.index >= cellN || c.Key() != in.Key
+	newEstPageSize := leafPageSize(leafPage)
+	if isInsert {
+		newEstPageSize += in.Size() + leafCellIndexElemSize
+	} else {
+		newEstPageSize += in.Size() - len(readLeafCellBytesAtOffset(leafPage, readCellOffset(leafPage, elem.index)))
+	}
+
+	// Use an optimized routine to insert the leaf cell if we won't overflow.
+	// We pad the estimate with 8 bytes because we do 8-byte alignment of cells.
+	if newEstPageSize+8 <= PageSize {
+		return c.putLeafCellFast(in, isInsert)
+	}
+
+	cells := readLeafCells(leafPage, c.leafCells[:])
+	cell := in
+	if isInsert {
 		//new cell
 		if in.Type == ContainerTypeBitmap {
 			//allocated bitmap()
@@ -476,6 +494,59 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
 	return c.putBranchCells(c.stack.index-1, parents)
 }
 
+// putLeafCellFast quickly insert or updates a cell on a leaf page.
+// It works by shifting bytes around instead of deserializing. This must not overflow.
+func (c *Cursor) putLeafCellFast(in leafCell, isInsert bool) (err error) {
+	src := c.leafPage
+	elem := &c.stack.elems[c.stack.index]
+	srcCellN := readCellN(src)
+
+	// Determine the cell count of the new page.
+	dstCellN := srcCellN
+	if isInsert {
+		dstCellN++
+	}
+
+	// Write page header.
+	dst := make([]byte, PageSize)
+	writePageNo(dst, readPageNo(src))
+	writeFlags(dst, PageTypeLeaf)
+	writeCellN(dst, dstCellN)
+
+	// Loop over source page elements and copy them to the new page.
+	offset := dataOffset(dstCellN)
+	for i, j := 0, 0; j < dstCellN; i, j = i+1, j+1 {
+		// If positioned at the insert/update index, write the new cell.
+		if i == elem.index {
+			writeLeafCell(dst[:], j, offset, in)
+			offset += align8(in.Size())
+
+			// If this is an update, skip to the next element.
+			if !isInsert {
+				continue
+			}
+
+			// If this is an insert, move the dst position forward.
+			j++
+		}
+
+		// Copy the raw bytes from the src page to the dst page.
+		if i < srcCellN {
+			srcCellBuf := readLeafCellBytesAtOffset(src, readCellOffset(src, i))
+			writeCellOffset(dst, j, offset)
+			copy(dst[offset:], srcCellBuf)
+			offset += align8(len(srcCellBuf))
+		}
+	}
+
+	// Write new page to dirty page cache.
+	if err := c.tx.writePage(dst); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // deleteLeafCell removes a cell from the currently positioned page & index.
 func (c *Cursor) deleteLeafCell(key uint64) (err error) {
 	cells := readLeafCells(c.leafPage, c.leafCells[:])

rbf/rbf.go

@@ -23,6 +23,8 @@ import (
 	"io"
 	"math"
 	"os"
+	"sync"
+	"time"
 	"unsafe"
 
 	"github.com/glycerine/rbtree"
@@ -99,6 +101,8 @@ const (
 	rootRecordPageHeaderSize = 12
 	rootRecordHeaderSize     = 4 + 2     // pgno, len(name)
 	leafCellHeaderSize       = 8 + 4 + 6 // key, type, count
+	leafPageHeaderSize       = 4 + 4 + 2 // pgno, flags, cell n
+	leafCellIndexElemSize    = 2
 	branchCellSize           = 8 + 4 + 4 // key, flags, pgno
 )
 
@@ -505,6 +509,35 @@ func readLeafCells(page []byte, buf []leafCell) []leafCell {
 	return cells
 }
 
+func readLeafCellBytesAtOffset(page []byte, offset int) []byte {
+	buf := page[offset:]
+	typ := ContainerType(*(*uint32)(unsafe.Pointer(&buf[8])))
+	n := int(*(*uint16)(unsafe.Pointer(&buf[12])))
+
+	switch typ {
+	case ContainerTypeArray:
+		return buf[:18+(n*2)]
+	case ContainerTypeRLE:
+		return buf[:18+(n*4)]
+	case ContainerTypeBitmapPtr:
+		return buf[:18+4]
+	default:
+		panic(fmt.Sprintf("invalid cell type: %d", typ))
+	}
+}
+
+// leafPageSize returns the number of bytes used on a leaf page.
+func leafPageSize(page []byte) int {
+	cellN := readCellN(page)
+	if cellN == 0 {
+		return leafPageHeaderSize
+	}
+
+	// Determine the offset & size of the last element.
+	offset := readCellOffset(page, cellN-1)
+	return offset + len(readLeafCellBytesAtOffset(page, offset))
+}
+
 // leafCellsPageSize returns the total page size required to hold cells.
 func leafCellsPageSize(cells []leafCell) int {
 	sz := dataOffset(len(cells))
@@ -712,3 +745,30 @@ func hashUint64(value uint64) uint32 {
 	}
 	return uint32(hash)
 }
+
+// Metric is a simple, internal metric for check duration of operations.
+type Metric struct {
+	name     string
+	interval int // reporting interval
+
+	mu sync.Mutex
+	d  time.Duration // total duration
+	n  int           // total count
+}
+
+func NewMetric(name string, interval int) Metric {
+	assert(interval > 0)
+	return Metric{name: name, interval: interval}
+}
+
+func (m *Metric) Inc(d time.Duration) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.d += d
+	m.n++
+
+	if m.n != 0 && m.n%m.interval == 0 {
+		fmt.Printf("metric:%10s avg=%dns\n", m.name, int(m.d)/m.n)
+	}
+}

rbf/tx_test.go

@@ -399,7 +399,7 @@ func setArray(tb testing.TB, key, num int, c *rbf.Cursor) {
 }
 
 func BenchmarkTx_Add(b *testing.B) {
-	for _, n := range []int{10000, 100000, 1000000} {
+	for _, n := range []int{1, 10, 1000} {
 		b.Run(fmt.Sprint(n), func(b *testing.B) {
 			rand := rand.New(rand.NewSource(0))
 
@@ -408,25 +408,27 @@ func BenchmarkTx_Add(b *testing.B) {
 				values[i] = uint64(rand.Intn(rbf.ShardWidth))
 			}
 			b.ResetTimer()
-			t := time.Now()
+
 			b.ReportAllocs()
 
 			for i := 0; i < b.N; i++ {
 				func() {
 					db := MustOpenDB(b)
 					defer MustCloseDB(b, db)
-					tx := MustBegin(b, db, true)
-					defer tx.Rollback()
 
 					for _, v := range values {
-						if _, err := tx.Add("x", v); err != nil {
-							b.Fatalf("Add(%d) i=%d err=%q", v, i, err)
-						}
+						func() {
+							tx := MustBegin(b, db, true)
+							defer tx.Rollback()
+							if _, err := tx.Add("x", v); err != nil {
+								b.Fatal(err)
+							} else if err := tx.Commit(); err != nil {
+								b.Fatal(err)
+							}
+						}()
 					}
 				}()
 			}
-
-			b.ReportMetric(float64(time.Since(t).Nanoseconds())/float64(n*b.N), "ns/op")
 		})
 	}
 }

rbf/util_test.go

@@ -41,8 +41,6 @@ func (c_orig *Cursor) DebugSlowCheckAllPages() {
 		}
 	}
 
-	c2.Dump("debug.dot")
-
 	checkElemNBitN(c2.tx, 0)
 }