Do index compaction in batches of 10000 in order to allow greater rea…

…d/write throughput when the index tree is large (greater than 1M entries)

mvcc/index.go

@@ -184,16 +184,40 @@ func (ti *treeIndex) Compact(rev int64) map[revision]struct{} {
 	available := make(map[revision]struct{})
 	var emptyki []*keyIndex
 	plog.Printf("store.index: compact %d", rev)
-	// TODO: do not hold the lock for long time?
-	// This is probably OK. Compacting 10M keys takes O(10ms).
-	ti.Lock()
-	defer ti.Unlock()
-	ti.tree.Ascend(compactIndex(rev, available, &emptyki))
-	for _, ki := range emptyki {
-		item := ti.tree.Delete(ki)
-		if item == nil {
-			plog.Panic("store.index: unexpected delete failure during compaction")
+	next := true
+	var keyi *keyIndex
+	minitem := ti.tree.Min()
+	if minitem != nil {
+		keyi = ti.tree.Min().(*keyIndex)
+	}
+
+	for next {
+	  i := 0
+
+	  ti.Lock()
+		next = false
+		ti.tree.AscendGreaterOrEqual(keyi, func(item btree.Item) bool {
+			i++
+			keyi = item.(*keyIndex)
+
+			if i > 10000 {
+				next = true
+				return false
+			}
+			keyi.compact(rev, available)
+			if keyi.isEmpty() {
+				emptyki = append(emptyki, keyi)
+			}
+			return true
+		})
+		for _, ki := range emptyki {
+			item := ti.tree.Delete(ki)
+			if item == nil {
+				plog.Panic("store.index: unexpected delete failure during compaction")
+			}
 		}
+		emptyki = emptyki[:0]
+		ti.Unlock()
 	}
 	return available
 }
@@ -211,17 +235,6 @@ func (ti *treeIndex) Keep(rev int64) map[revision]struct{} {
 	return available
 }
 
-func compactIndex(rev int64, available map[revision]struct{}, emptyki *[]*keyIndex) func(i btree.Item) bool {
-	return func(i btree.Item) bool {
-		keyi := i.(*keyIndex)
-		keyi.compact(rev, available)
-		if keyi.isEmpty() {
-			*emptyki = append(*emptyki, keyi)
-		}
-		return true
-	}
-}
-
 func (ti *treeIndex) Equal(bi index) bool {
 	b := bi.(*treeIndex)
 

mvcc/index_bench_test.go

@@ -0,0 +1,26 @@
+package mvcc
+
+import (
+	"testing"
+)
+
+func BenchmarkIndexCompact1(b *testing.B)   { benchmarkIndexCompact(b, 1) }
+func BenchmarkIndexCompact100(b *testing.B) { benchmarkIndexCompact(b, 100) }
+func BenchmarkIndexCompact10000(b *testing.B) { benchmarkIndexCompact(b, 10000) }
+func BenchmarkIndexCompact100000(b *testing.B) { benchmarkIndexCompact(b, 100000) }
+func BenchmarkIndexCompact1000000(b *testing.B) { benchmarkIndexCompact(b, 1000000) }
+
+func benchmarkIndexCompact(b *testing.B, size int) {
+	plog.SetLevel(0) // suppress log entries
+  kvindex := newTreeIndex()
+
+	bytesN := 64
+	keys := createBytesSlice(bytesN, size)
+	for i := 1; i < size; i++ {
+		kvindex.Put(keys[i], revision {main: int64(i), sub: int64(i)})
+	}
+	b.ResetTimer()
+	for i := 1; i < b.N; i++ {
+		kvindex.Compact(int64(i))
+	}
+}